Repository: prometheus/tsdb
Pull request #526 (2019-02-15 18:27:26)

Merge encoding_helpers.go to tsdbutil

bwplotka: Hey, what is the reason behind this? (: 

`util` in Golang packages are antipattern as described here: https://blog.golang.org/package-names#TOC_5 This is because it usually ends up being pile of mixed functions from everywhere and makes reading the code more confusing.
naivewong: Oh sorry, I forgot to tag this PR. This PR would refer to #523.

And in this case, since the tsdbutil package already existed and the encoding_helpers.go contains two general tools (encbuf, decbuf) which are defined and used seperately in packcage index and tsdb(and maybe possible in the future packages), then I think maybe it's resonable to merge the duplicates to tsdbutil.
krasi-georgiev: @bwplotka 	good point. Maybe a better option is to create a package called `encoding`.
naivewong: @krasi-georgiev But for the current changes, is it ok to you? I may do the changes all together later
krasi-georgiev: yes I would say move all relevant funcs inside a package called `encoding` or some other more relevant name.

Patch:
diff --git a/.travis.yml b/.travis.yml
index c5012a3..faed514 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -20,5 +20,4 @@ install:
   - make deps
 
 script:
-  # `staticcheck` target is omitted due to linting errors
-  - if [[ "$TRAVIS_OS_NAME" == "windows" ]]; then make test; else make; fi
+  - if [[ "$TRAVIS_OS_NAME" == "windows" ]]; then make test; else make all; fi
diff --git a/Makefile.common b/Makefile.common
index a422e1b..7e86f45 100644
--- a/Makefile.common
+++ b/Makefile.common
@@ -87,14 +87,14 @@ ifeq ($(GOHOSTARCH),amd64)
         endif
 endif
 
-.PHONY: all
-all: precheck style staticcheck unused build test
-
 # This rule is used to forward a target like "build" to "common-build".  This
 # allows a new "build" target to be defined in a Makefile which includes this
 # one and override "common-build" without override warnings.
 %: common-% ;
 
+.PHONY: common-all
+common-all: precheck style check_license staticcheck unused build test
+
 .PHONY: common-style
 common-style:
 	@echo ">> checking code style"
diff --git a/encoding/encoding.go b/encoding/encoding.go
new file mode 100644
index 0000000..395e11c
--- /dev/null
+++ b/encoding/encoding.go
@@ -0,0 +1,244 @@
+// Copyright 2018 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 encoding
+
+import (
+	"encoding/binary"
+	"hash"
+	"hash/crc32"
+	"unsafe"
+
+	"github.com/pkg/errors"
+)
+
+var (
+	ErrInvalidSize     = errors.New("invalid size")
+	ErrInvalidChecksum = errors.New("invalid checksum")
+)
+
+// enbuf is a helper type to populate a byte slice with various types.
+type Encbuf struct {
+	B []byte
+	C [binary.MaxVarintLen64]byte
+}
+
+func (e *Encbuf) Reset()      { e.B = e.B[:0] }
+func (e *Encbuf) Get() []byte { return e.B }
+func (e *Encbuf) Len() int    { return len(e.B) }
+
+func (e *Encbuf) PutString(s string) { e.B = append(e.B, s...) }
+func (e *Encbuf) PutByte(c byte)     { e.B = append(e.B, c) }
+
+func (e *Encbuf) PutBE32int(x int)      { e.PutBE32(uint32(x)) }
+func (e *Encbuf) PutUvarint32(x uint32) { e.PutUvarint64(uint64(x)) }
+func (e *Encbuf) PutBE64int64(x int64)  { e.PutBE64(uint64(x)) }
+func (e *Encbuf) PutUvarint(x int)      { e.PutUvarint64(uint64(x)) }
+
+func (e *Encbuf) PutBE32(x uint32) {
+	binary.BigEndian.PutUint32(e.C[:], x)
+	e.B = append(e.B, e.C[:4]...)
+}
+
+func (e *Encbuf) PutBE64(x uint64) {
+	binary.BigEndian.PutUint64(e.C[:], x)
+	e.B = append(e.B, e.C[:8]...)
+}
+
+func (e *Encbuf) PutUvarint64(x uint64) {
+	n := binary.PutUvarint(e.C[:], x)
+	e.B = append(e.B, e.C[:n]...)
+}
+
+func (e *Encbuf) PutVarint64(x int64) {
+	n := binary.PutVarint(e.C[:], x)
+	e.B = append(e.B, e.C[:n]...)
+}
+
+// putVarintStr writes a string to the buffer prefixed by its varint length (in bytes!).
+func (e *Encbuf) PutUvarintStr(s string) {
+	b := *(*[]byte)(unsafe.Pointer(&s))
+	e.PutUvarint(len(b))
+	e.PutString(s)
+}
+
+// putHash appends a hash over the buffers current contents to the buffer.
+func (e *Encbuf) PutHash(h hash.Hash) {
+	h.Reset()
+	_, err := h.Write(e.B)
+	if err != nil {
+		panic(err) // The CRC32 implementation does not error
+	}
+	e.B = h.Sum(e.B)
+}
+
+// decbuf provides safe methods to extract data from a byte slice. It does all
+// necessary bounds checking and advancing of the byte slice.
+// Several datums can be extracted without checking for errors. However, before using
+// any datum, the err() method must be checked.
+type Decbuf struct {
+	B []byte
+	E error
+}
+
+// NewDecbufAt returns a new decoding buffer. It expects the first 4 bytes
+// after offset to hold the big endian encoded content length, followed by the contents and the expected
+// checksum.
+func NewDecbufAt(bs ByteSlice, off int, castagnoliTable *crc32.Table) Decbuf {
+	if bs.Len() < off+4 {
+		return Decbuf{E: ErrInvalidSize}
+	}
+	b := bs.Range(off, off+4)
+	l := int(binary.BigEndian.Uint32(b))
+
+	if bs.Len() < off+4+l+4 {
+		return Decbuf{E: ErrInvalidSize}
+	}
+
+	// Load bytes holding the contents plus a CRC32 checksum.
+	b = bs.Range(off+4, off+4+l+4)
+	dec := Decbuf{B: b[:len(b)-4]}
+
+	if exp := binary.BigEndian.Uint32(b[len(b)-4:]); dec.Crc32(castagnoliTable) != exp {
+		return Decbuf{E: ErrInvalidChecksum}
+	}
+	return dec
+}
+
+// NewDecbufUvarintAt returns a new decoding buffer. It expects the first bytes
+// after offset to hold the uvarint-encoded buffers length, followed by the contents and the expected
+// checksum.
+func NewDecbufUvarintAt(bs ByteSlice, off int, castagnoliTable *crc32.Table) Decbuf {
+	// We never have to access this method at the far end of the byte slice. Thus just checking
+	// against the MaxVarintLen32 is sufficient.
+	if bs.Len() < off+binary.MaxVarintLen32 {
+		return Decbuf{E: ErrInvalidSize}
+	}
+	b := bs.Range(off, off+binary.MaxVarintLen32)
+
+	l, n := binary.Uvarint(b)
+	if n <= 0 || n > binary.MaxVarintLen32 {
+		return Decbuf{E: errors.Errorf("invalid uvarint %d", n)}
+	}
+
+	if bs.Len() < off+n+int(l)+4 {
+		return Decbuf{E: ErrInvalidSize}
+	}
+
+	// Load bytes holding the contents plus a CRC32 checksum.
+	b = bs.Range(off+n, off+n+int(l)+4)
+	dec := Decbuf{B: b[:len(b)-4]}
+
+	if dec.Crc32(castagnoliTable) != binary.BigEndian.Uint32(b[len(b)-4:]) {
+		return Decbuf{E: ErrInvalidChecksum}
+	}
+	return dec
+}
+
+func (d *Decbuf) Uvarint() int     { return int(d.Uvarint64()) }
+func (d *Decbuf) Be32int() int     { return int(d.Be32()) }
+func (d *Decbuf) Be64int64() int64 { return int64(d.Be64()) }
+
+// Crc32 returns a CRC32 checksum over the remaining bytes.
+func (d *Decbuf) Crc32(castagnoliTable *crc32.Table) uint32 {
+	return crc32.Checksum(d.B, castagnoliTable)
+}
+
+func (d *Decbuf) UvarintStr() string {
+	l := d.Uvarint64()
+	if d.E != nil {
+		return ""
+	}
+	if len(d.B) < int(l) {
+		d.E = ErrInvalidSize
+		return ""
+	}
+	s := string(d.B[:l])
+	d.B = d.B[l:]
+	return s
+}
+
+func (d *Decbuf) Varint64() int64 {
+	if d.E != nil {
+		return 0
+	}
+	x, n := binary.Varint(d.B)
+	if n < 1 {
+		d.E = ErrInvalidSize
+		return 0
+	}
+	d.B = d.B[n:]
+	return x
+}
+
+func (d *Decbuf) Uvarint64() uint64 {
+	if d.E != nil {
+		return 0
+	}
+	x, n := binary.Uvarint(d.B)
+	if n < 1 {
+		d.E = ErrInvalidSize
+		return 0
+	}
+	d.B = d.B[n:]
+	return x
+}
+
+func (d *Decbuf) Be64() uint64 {
+	if d.E != nil {
+		return 0
+	}
+	if len(d.B) < 8 {
+		d.E = ErrInvalidSize
+		return 0
+	}
+	x := binary.BigEndian.Uint64(d.B)
+	d.B = d.B[8:]
+	return x
+}
+
+func (d *Decbuf) Be32() uint32 {
+	if d.E != nil {
+		return 0
+	}
+	if len(d.B) < 4 {
+		d.E = ErrInvalidSize
+		return 0
+	}
+	x := binary.BigEndian.Uint32(d.B)
+	d.B = d.B[4:]
+	return x
+}
+
+func (d *Decbuf) Byte() byte {
+	if d.E != nil {
+		return 0
+	}
+	if len(d.B) < 1 {
+		d.E = ErrInvalidSize
+		return 0
+	}
+	x := d.B[0]
+	d.B = d.B[1:]
+	return x
+}
+
+func (d *Decbuf) Err() error  { return d.E }
+func (d *Decbuf) Len() int    { return len(d.B) }
+func (d *Decbuf) Get() []byte { return d.B }
+
+// ByteSlice abstracts a byte slice.
+type ByteSlice interface {
+	Len() int
+	Range(start, end int) []byte
+}
diff --git a/encoding_helpers.go b/encoding_helpers.go
deleted file mode 100644
index 9c10e31..0000000
--- a/encoding_helpers.go
+++ /dev/null
@@ -1,160 +0,0 @@
-// Copyright 2018 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 (
-	"encoding/binary"
-	"unsafe"
-
-	"github.com/pkg/errors"
-)
-
-var errInvalidSize = errors.New("invalid size")
-
-// encbuf is a helper type to populate a byte slice with various types.
-type encbuf struct {
-	b []byte
-	c [binary.MaxVarintLen64]byte
-}
-
-func (e *encbuf) reset()      { e.b = e.b[:0] }
-func (e *encbuf) get() []byte { return e.b }
-
-func (e *encbuf) putString(s string) { e.b = append(e.b, s...) }
-func (e *encbuf) putByte(c byte)     { e.b = append(e.b, c) }
-
-func (e *encbuf) putBE64int64(x int64) { e.putBE64(uint64(x)) }
-func (e *encbuf) putUvarint(x int)     { e.putUvarint64(uint64(x)) }
-
-func (e *encbuf) putBE32(x uint32) {
-	binary.BigEndian.PutUint32(e.c[:], x)
-	e.b = append(e.b, e.c[:4]...)
-}
-
-func (e *encbuf) putBE64(x uint64) {
-	binary.BigEndian.PutUint64(e.c[:], x)
-	e.b = append(e.b, e.c[:8]...)
-}
-
-func (e *encbuf) putUvarint64(x uint64) {
-	n := binary.PutUvarint(e.c[:], x)
-	e.b = append(e.b, e.c[:n]...)
-}
-
-func (e *encbuf) putVarint64(x int64) {
-	n := binary.PutVarint(e.c[:], x)
-	e.b = append(e.b, e.c[:n]...)
-}
-
-// putVarintStr writes a string to the buffer prefixed by its varint length (in bytes!).
-func (e *encbuf) putUvarintStr(s string) {
-	b := *(*[]byte)(unsafe.Pointer(&s))
-	e.putUvarint(len(b))
-	e.putString(s)
-}
-
-// decbuf provides safe methods to extract data from a byte slice. It does all
-// necessary bounds checking and advancing of the byte slice.
-// Several datums can be extracted without checking for errors. However, before using
-// any datum, the err() method must be checked.
-type decbuf struct {
-	b []byte
-	e error
-}
-
-func (d *decbuf) uvarint() int     { return int(d.uvarint64()) }
-func (d *decbuf) be64int64() int64 { return int64(d.be64()) }
-
-func (d *decbuf) uvarintStr() string {
-	l := d.uvarint64()
-	if d.e != nil {
-		return ""
-	}
-	if len(d.b) < int(l) {
-		d.e = errInvalidSize
-		return ""
-	}
-	s := string(d.b[:l])
-	d.b = d.b[l:]
-	return s
-}
-
-func (d *decbuf) varint64() int64 {
-	if d.e != nil {
-		return 0
-	}
-	x, n := binary.Varint(d.b)
-	if n < 1 {
-		d.e = errInvalidSize
-		return 0
-	}
-	d.b = d.b[n:]
-	return x
-}
-
-func (d *decbuf) uvarint64() uint64 {
-	if d.e != nil {
-		return 0
-	}
-	x, n := binary.Uvarint(d.b)
-	if n < 1 {
-		d.e = errInvalidSize
-		return 0
-	}
-	d.b = d.b[n:]
-	return x
-}
-
-func (d *decbuf) be64() uint64 {
-	if d.e != nil {
-		return 0
-	}
-	if len(d.b) < 4 {
-		d.e = errInvalidSize
-		return 0
-	}
-	x := binary.BigEndian.Uint64(d.b)
-	d.b = d.b[8:]
-	return x
-}
-
-func (d *decbuf) be32() uint32 {
-	if d.e != nil {
-		return 0
-	}
-	if len(d.b) < 4 {
-		d.e = errInvalidSize
-		return 0
-	}
-	x := binary.BigEndian.Uint32(d.b)
-	d.b = d.b[4:]
-	return x
-}
-
-func (d *decbuf) byte() byte {
-	if d.e != nil {
-		return 0
-	}
-	if len(d.b) < 1 {
-		d.e = errInvalidSize
-		return 0
-	}
-	x := d.b[0]
-	d.b = d.b[1:]
-	return x
-}
-
-func (d *decbuf) err() error  { return d.e }
-func (d *decbuf) len() int    { return len(d.b) }
-func (d *decbuf) get() []byte { return d.b }
diff --git a/head.go b/head.go
index c5ac06c..fe20897 100644
--- a/head.go
+++ b/head.go
@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/tsdb/chunks"
+	"github.com/prometheus/tsdb/encoding"
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/wal"
@@ -1117,7 +1118,7 @@ func (h *headIndexReader) Symbols() (map[string]struct{}, error) {
 // LabelValues returns the possible label values
 func (h *headIndexReader) LabelValues(names ...string) (index.StringTuples, error) {
 	if len(names) != 1 {
-		return nil, errInvalidSize
+		return nil, encoding.ErrInvalidSize
 	}
 
 	h.head.symMtx.RLock()
diff --git a/index/encoding_helpers.go b/index/encoding_helpers.go
deleted file mode 100644
index 1ed1301..0000000
--- a/index/encoding_helpers.go
+++ /dev/null
@@ -1,218 +0,0 @@
-// Copyright 2018 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 index
-
-import (
-	"encoding/binary"
-	"hash"
-	"hash/crc32"
-	"unsafe"
-
-	"github.com/pkg/errors"
-)
-
-// enbuf is a helper type to populate a byte slice with various types.
-type encbuf struct {
-	b []byte
-	c [binary.MaxVarintLen64]byte
-}
-
-func (e *encbuf) reset()      { e.b = e.b[:0] }
-func (e *encbuf) get() []byte { return e.b }
-func (e *encbuf) len() int    { return len(e.b) }
-
-func (e *encbuf) putString(s string) { e.b = append(e.b, s...) }
-func (e *encbuf) putByte(c byte)     { e.b = append(e.b, c) }
-
-func (e *encbuf) putBE32int(x int)      { e.putBE32(uint32(x)) }
-func (e *encbuf) putUvarint32(x uint32) { e.putUvarint64(uint64(x)) }
-func (e *encbuf) putUvarint(x int)      { e.putUvarint64(uint64(x)) }
-
-func (e *encbuf) putBE32(x uint32) {
-	binary.BigEndian.PutUint32(e.c[:], x)
-	e.b = append(e.b, e.c[:4]...)
-}
-
-func (e *encbuf) putBE64(x uint64) {
-	binary.BigEndian.PutUint64(e.c[:], x)
-	e.b = append(e.b, e.c[:8]...)
-}
-
-func (e *encbuf) putUvarint64(x uint64) {
-	n := binary.PutUvarint(e.c[:], x)
-	e.b = append(e.b, e.c[:n]...)
-}
-
-func (e *encbuf) putVarint64(x int64) {
-	n := binary.PutVarint(e.c[:], x)
-	e.b = append(e.b, e.c[:n]...)
-}
-
-// putVarintStr writes a string to the buffer prefixed by its varint length (in bytes!).
-func (e *encbuf) putUvarintStr(s string) {
-	b := *(*[]byte)(unsafe.Pointer(&s))
-	e.putUvarint(len(b))
-	e.putString(s)
-}
-
-// putHash appends a hash over the buffers current contents to the buffer.
-func (e *encbuf) putHash(h hash.Hash) {
-	h.Reset()
-	_, err := h.Write(e.b)
-	if err != nil {
-		panic(err) // The CRC32 implementation does not error
-	}
-	e.b = h.Sum(e.b)
-}
-
-// decbuf provides safe methods to extract data from a byte slice. It does all
-// necessary bounds checking and advancing of the byte slice.
-// Several datums can be extracted without checking for errors. However, before using
-// any datum, the err() method must be checked.
-type decbuf struct {
-	b []byte
-	e error
-}
-
-// newDecbufAt returns a new decoding buffer. It expects the first 4 bytes
-// after offset to hold the big endian encoded content length, followed by the contents and the expected
-// checksum.
-func newDecbufAt(bs ByteSlice, off int) decbuf {
-	if bs.Len() < off+4 {
-		return decbuf{e: errInvalidSize}
-	}
-	b := bs.Range(off, off+4)
-	l := int(binary.BigEndian.Uint32(b))
-
-	if bs.Len() < off+4+l+4 {
-		return decbuf{e: errInvalidSize}
-	}
-
-	// Load bytes holding the contents plus a CRC32 checksum.
-	b = bs.Range(off+4, off+4+l+4)
-	dec := decbuf{b: b[:len(b)-4]}
-
-	if exp := binary.BigEndian.Uint32(b[len(b)-4:]); dec.crc32() != exp {
-		return decbuf{e: errInvalidChecksum}
-	}
-	return dec
-}
-
-// decbufUvarintAt returns a new decoding buffer. It expects the first bytes
-// after offset to hold the uvarint-encoded buffers length, followed by the contents and the expected
-// checksum.
-func newDecbufUvarintAt(bs ByteSlice, off int) decbuf {
-	// We never have to access this method at the far end of the byte slice. Thus just checking
-	// against the MaxVarintLen32 is sufficient.
-	if bs.Len() < off+binary.MaxVarintLen32 {
-		return decbuf{e: errInvalidSize}
-	}
-	b := bs.Range(off, off+binary.MaxVarintLen32)
-
-	l, n := binary.Uvarint(b)
-	if n <= 0 || n > binary.MaxVarintLen32 {
-		return decbuf{e: errors.Errorf("invalid uvarint %d", n)}
-	}
-
-	if bs.Len() < off+n+int(l)+4 {
-		return decbuf{e: errInvalidSize}
-	}
-
-	// Load bytes holding the contents plus a CRC32 checksum.
-	b = bs.Range(off+n, off+n+int(l)+4)
-	dec := decbuf{b: b[:len(b)-4]}
-
-	if dec.crc32() != binary.BigEndian.Uint32(b[len(b)-4:]) {
-		return decbuf{e: errInvalidChecksum}
-	}
-	return dec
-}
-
-func (d *decbuf) uvarint() int { return int(d.uvarint64()) }
-func (d *decbuf) be32int() int { return int(d.be32()) }
-
-// crc32 returns a CRC32 checksum over the remaining bytes.
-func (d *decbuf) crc32() uint32 {
-	return crc32.Checksum(d.b, castagnoliTable)
-}
-
-func (d *decbuf) uvarintStr() string {
-	l := d.uvarint64()
-	if d.e != nil {
-		return ""
-	}
-	if len(d.b) < int(l) {
-		d.e = errInvalidSize
-		return ""
-	}
-	s := string(d.b[:l])
-	d.b = d.b[l:]
-	return s
-}
-
-func (d *decbuf) varint64() int64 {
-	if d.e != nil {
-		return 0
-	}
-	x, n := binary.Varint(d.b)
-	if n < 1 {
-		d.e = errInvalidSize
-		return 0
-	}
-	d.b = d.b[n:]
-	return x
-}
-
-func (d *decbuf) uvarint64() uint64 {
-	if d.e != nil {
-		return 0
-	}
-	x, n := binary.Uvarint(d.b)
-	if n < 1 {
-		d.e = errInvalidSize
-		return 0
-	}
-	d.b = d.b[n:]
-	return x
-}
-
-func (d *decbuf) be64() uint64 {
-	if d.e != nil {
-		return 0
-	}
-	if len(d.b) < 8 {
-		d.e = errInvalidSize
-		return 0
-	}
-	x := binary.BigEndian.Uint64(d.b)
-	d.b = d.b[8:]
-	return x
-}
-
-func (d *decbuf) be32() uint32 {
-	if d.e != nil {
-		return 0
-	}
-	if len(d.b) < 4 {
-		d.e = errInvalidSize
-		return 0
-	}
-	x := binary.BigEndian.Uint32(d.b)
-	d.b = d.b[4:]
-	return x
-}
-
-func (d *decbuf) err() error  { return d.e }
-func (d *decbuf) len() int    { return len(d.b) }
-func (d *decbuf) get() []byte { return d.b }
diff --git a/index/index.go b/index/index.go
index 442e025..580291a 100644
--- a/index/index.go
+++ b/index/index.go
@@ -16,7 +16,6 @@ package index
 import (
 	"bufio"
 	"encoding/binary"
-	"fmt"
 	"hash"
 	"hash/crc32"
 	"io"
@@ -29,6 +28,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb/chunks"
+	"github.com/prometheus/tsdb/encoding"
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/labels"
 )
@@ -119,8 +119,8 @@ type Writer struct {
 	stage indexWriterStage
 
 	// Reusable memory.
-	buf1    encbuf
-	buf2    encbuf
+	buf1    encoding.Encbuf
+	buf2    encoding.Encbuf
 	uint32s []uint32
 
 	symbols       map[string]uint32 // symbol offsets
@@ -149,28 +149,28 @@ type TOC struct {
 // NewTOCFromByteSlice return parsed TOC from given index byte slice.
 func NewTOCFromByteSlice(bs ByteSlice) (*TOC, error) {
 	if bs.Len() < indexTOCLen {
-		return nil, errInvalidSize
+		return nil, encoding.ErrInvalidSize
 	}
 	b := bs.Range(bs.Len()-indexTOCLen, bs.Len())
 
 	expCRC := binary.BigEndian.Uint32(b[len(b)-4:])
-	d := decbuf{b: b[:len(b)-4]}
+	d := encoding.Decbuf{B: b[:len(b)-4]}
 
-	if d.crc32() != expCRC {
-		return nil, errors.Wrap(errInvalidChecksum, "read TOC")
+	if d.Crc32(castagnoliTable) != expCRC {
+		return nil, errors.Wrap(encoding.ErrInvalidChecksum, "read TOC")
 	}
 
-	if err := d.err(); err != nil {
+	if err := d.Err(); err != nil {
 		return nil, err
 	}
 
 	return &TOC{
-		Symbols:           d.be64(),
-		Series:            d.be64(),
-		LabelIndices:      d.be64(),
-		LabelIndicesTable: d.be64(),
-		Postings:          d.be64(),
-		PostingsTable:     d.be64(),
+		Symbols:           d.Be64(),
+		Series:            d.Be64(),
+		LabelIndices:      d.Be64(),
+		LabelIndicesTable: d.Be64(),
+		Postings:          d.Be64(),
+		PostingsTable:     d.Be64(),
 	}, nil
 }
 
@@ -203,8 +203,8 @@ func NewWriter(fn string) (*Writer, error) {
 		stage: idxStageNone,
 
 		// Reusable memory.
-		buf1:    encbuf{b: make([]byte, 0, 1<<22)},
-		buf2:    encbuf{b: make([]byte, 0, 1<<22)},
+		buf1:    encoding.Encbuf{B: make([]byte, 0, 1<<22)},
+		buf2:    encoding.Encbuf{B: make([]byte, 0, 1<<22)},
 		uint32s: make([]uint32, 0, 1<<15),
 
 		// Caches.
@@ -288,11 +288,11 @@ func (w *Writer) ensureStage(s indexWriterStage) error {
 }
 
 func (w *Writer) writeMeta() error {
-	w.buf1.reset()
-	w.buf1.putBE32(MagicIndex)
-	w.buf1.putByte(FormatV2)
+	w.buf1.Reset()
+	w.buf1.PutBE32(MagicIndex)
+	w.buf1.PutByte(FormatV2)
 
-	return w.write(w.buf1.get())
+	return w.write(w.buf1.Get())
 }
 
 // AddSeries adds the series one at a time along with its chunks.
@@ -318,8 +318,8 @@ func (w *Writer) AddSeries(ref uint64, lset labels.Labels, chunks ...chunks.Meta
 	}
 	w.seriesOffsets[ref] = w.pos / 16
 
-	w.buf2.reset()
-	w.buf2.putUvarint(len(lset))
+	w.buf2.Reset()
+	w.buf2.PutUvarint(len(lset))
 
 	for _, l := range lset {
 		// here we have an index for the symbol file if v2, otherwise it's an offset
@@ -327,41 +327,41 @@ func (w *Writer) AddSeries(ref uint64, lset labels.Labels, chunks ...chunks.Meta
 		if !ok {
 			return errors.Errorf("symbol entry for %q does not exist", l.Name)
 		}
-		w.buf2.putUvarint32(index)
+		w.buf2.PutUvarint32(index)
 
 		index, ok = w.symbols[l.Value]
 		if !ok {
 			return errors.Errorf("symbol entry for %q does not exist", l.Value)
 		}
-		w.buf2.putUvarint32(index)
+		w.buf2.PutUvarint32(index)
 	}
 
-	w.buf2.putUvarint(len(chunks))
+	w.buf2.PutUvarint(len(chunks))
 
 	if len(chunks) > 0 {
 		c := chunks[0]
-		w.buf2.putVarint64(c.MinTime)
-		w.buf2.putUvarint64(uint64(c.MaxTime - c.MinTime))
-		w.buf2.putUvarint64(c.Ref)
+		w.buf2.PutVarint64(c.MinTime)
+		w.buf2.PutUvarint64(uint64(c.MaxTime - c.MinTime))
+		w.buf2.PutUvarint64(c.Ref)
 		t0 := c.MaxTime
 		ref0 := int64(c.Ref)
 
 		for _, c := range chunks[1:] {
-			w.buf2.putUvarint64(uint64(c.MinTime - t0))
-			w.buf2.putUvarint64(uint64(c.MaxTime - c.MinTime))
+			w.buf2.PutUvarint64(uint64(c.MinTime - t0))
+			w.buf2.PutUvarint64(uint64(c.MaxTime - c.MinTime))
 			t0 = c.MaxTime
 
-			w.buf2.putVarint64(int64(c.Ref) - ref0)
+			w.buf2.PutVarint64(int64(c.Ref) - ref0)
 			ref0 = int64(c.Ref)
 		}
 	}
 
-	w.buf1.reset()
-	w.buf1.putUvarint(w.buf2.len())
+	w.buf1.Reset()
+	w.buf1.PutUvarint(w.buf2.Len())
 
-	w.buf2.putHash(w.crc32)
+	w.buf2.PutHash(w.crc32)
 
-	if err := w.write(w.buf1.get(), w.buf2.get()); err != nil {
+	if err := w.write(w.buf1.Get(), w.buf2.Get()); err != nil {
 		return errors.Wrap(err, "write series data")
 	}
 
@@ -382,22 +382,22 @@ func (w *Writer) AddSymbols(sym map[string]struct{}) error {
 	}
 	sort.Strings(symbols)
 
-	w.buf1.reset()
-	w.buf2.reset()
+	w.buf1.Reset()
+	w.buf2.Reset()
 
-	w.buf2.putBE32int(len(symbols))
+	w.buf2.PutBE32int(len(symbols))
 
 	w.symbols = make(map[string]uint32, len(symbols))
 
 	for index, s := range symbols {
 		w.symbols[s] = uint32(index)
-		w.buf2.putUvarintStr(s)
+		w.buf2.PutUvarintStr(s)
 	}
 
-	w.buf1.putBE32int(w.buf2.len())
-	w.buf2.putHash(w.crc32)
+	w.buf1.PutBE32int(w.buf2.Len())
+	w.buf2.PutHash(w.crc32)
 
-	err := w.write(w.buf1.get(), w.buf2.get())
+	err := w.write(w.buf1.Get(), w.buf2.Get())
 	return errors.Wrap(err, "write symbols")
 }
 
@@ -425,9 +425,9 @@ func (w *Writer) WriteLabelIndex(names []string, values []string) error {
 		offset: w.pos,
 	})
 
-	w.buf2.reset()
-	w.buf2.putBE32int(len(names))
-	w.buf2.putBE32int(valt.Len())
+	w.buf2.Reset()
+	w.buf2.PutBE32int(len(names))
+	w.buf2.PutBE32int(valt.Len())
 
 	// here we have an index for the symbol file if v2, otherwise it's an offset
 	for _, v := range valt.entries {
@@ -435,53 +435,53 @@ func (w *Writer) WriteLabelIndex(names []string, values []string) error {
 		if !ok {
 			return errors.Errorf("symbol entry for %q does not exist", v)
 		}
-		w.buf2.putBE32(index)
+		w.buf2.PutBE32(index)
 	}
 
-	w.buf1.reset()
-	w.buf1.putBE32int(w.buf2.len())
+	w.buf1.Reset()
+	w.buf1.PutBE32int(w.buf2.Len())
 
-	w.buf2.putHash(w.crc32)
+	w.buf2.PutHash(w.crc32)
 
-	err = w.write(w.buf1.get(), w.buf2.get())
+	err = w.write(w.buf1.Get(), w.buf2.Get())
 	return errors.Wrap(err, "write label index")
 }
 
 // writeOffsetTable writes a sequence of readable hash entries.
 func (w *Writer) writeOffsetTable(entries []hashEntry) error {
-	w.buf2.reset()
-	w.buf2.putBE32int(len(entries))
+	w.buf2.Reset()
+	w.buf2.PutBE32int(len(entries))
 
 	for _, e := range entries {
-		w.buf2.putUvarint(len(e.keys))
+		w.buf2.PutUvarint(len(e.keys))
 		for _, k := range e.keys {
-			w.buf2.putUvarintStr(k)
+			w.buf2.PutUvarintStr(k)
 		}
-		w.buf2.putUvarint64(e.offset)
+		w.buf2.PutUvarint64(e.offset)
 	}
 
-	w.buf1.reset()
-	w.buf1.putBE32int(w.buf2.len())
-	w.buf2.putHash(w.crc32)
+	w.buf1.Reset()
+	w.buf1.PutBE32int(w.buf2.Len())
+	w.buf2.PutHash(w.crc32)
 
-	return w.write(w.buf1.get(), w.buf2.get())
+	return w.write(w.buf1.Get(), w.buf2.Get())
 }
 
 const indexTOCLen = 6*8 + 4
 
 func (w *Writer) writeTOC() error {
-	w.buf1.reset()
+	w.buf1.Reset()
 
-	w.buf1.putBE64(w.toc.Symbols)
-	w.buf1.putBE64(w.toc.Series)
-	w.buf1.putBE64(w.toc.LabelIndices)
-	w.buf1.putBE64(w.toc.LabelIndicesTable)
-	w.buf1.putBE64(w.toc.Postings)
-	w.buf1.putBE64(w.toc.PostingsTable)
+	w.buf1.PutBE64(w.toc.Symbols)
+	w.buf1.PutBE64(w.toc.Series)
+	w.buf1.PutBE64(w.toc.LabelIndices)
+	w.buf1.PutBE64(w.toc.LabelIndicesTable)
+	w.buf1.PutBE64(w.toc.Postings)
+	w.buf1.PutBE64(w.toc.PostingsTable)
 
-	w.buf1.putHash(w.crc32)
+	w.buf1.PutHash(w.crc32)
 
-	return w.write(w.buf1.get())
+	return w.write(w.buf1.Get())
 }
 
 func (w *Writer) WritePostings(name, value string, it Postings) error {
@@ -519,20 +519,20 @@ func (w *Writer) WritePostings(name, value string, it Postings) error {
 	}
 	sort.Sort(uint32slice(refs))
 
-	w.buf2.reset()
-	w.buf2.putBE32int(len(refs))
+	w.buf2.Reset()
+	w.buf2.PutBE32int(len(refs))
 
 	for _, r := range refs {
-		w.buf2.putBE32(r)
+		w.buf2.PutBE32(r)
 	}
 	w.uint32s = refs
 
-	w.buf1.reset()
-	w.buf1.putBE32int(w.buf2.len())
+	w.buf1.Reset()
+	w.buf1.PutBE32int(w.buf2.Len())
 
-	w.buf2.putHash(w.crc32)
+	w.buf2.PutHash(w.crc32)
 
-	err := w.write(w.buf1.get(), w.buf2.get())
+	err := w.write(w.buf1.Get(), w.buf2.Get())
 	return errors.Wrap(err, "write postings")
 }
 
@@ -593,11 +593,6 @@ type Reader struct {
 	version int
 }
 
-var (
-	errInvalidSize     = fmt.Errorf("invalid size")
-	errInvalidChecksum = fmt.Errorf("invalid checksum")
-)
-
 // ByteSlice abstracts a byte slice.
 type ByteSlice interface {
 	Len() int
@@ -643,7 +638,7 @@ func newReader(b ByteSlice, c io.Closer) (*Reader, error) {
 
 	// Verify header.
 	if r.b.Len() < HeaderLen {
-		return nil, errors.Wrap(errInvalidSize, "index header")
+		return nil, errors.Wrap(encoding.ErrInvalidSize, "index header")
 	}
 	if m := binary.BigEndian.Uint32(r.b.Range(0, 4)); m != MagicIndex {
 		return nil, errors.Errorf("invalid magic number %x", m)
@@ -724,13 +719,13 @@ func (r *Reader) PostingsRanges() (map[labels.Label]Range, error) {
 
 	for k, e := range r.postings {
 		for v, start := range e {
-			d := newDecbufAt(r.b, int(start))
-			if d.err() != nil {
-				return nil, d.err()
+			d := encoding.NewDecbufAt(r.b, int(start), castagnoliTable)
+			if d.Err() != nil {
+				return nil, d.Err()
 			}
 			m[labels.Label{Name: k, Value: v}] = Range{
 				Start: int64(start) + 4,
-				End:   int64(start) + 4 + int64(d.len()),
+				End:   int64(start) + 4 + int64(d.Len()),
 			}
 		}
 	}
@@ -744,13 +739,13 @@ func ReadSymbols(bs ByteSlice, version int, off int) ([]string, map[uint32]strin
 	if off == 0 {
 		return nil, nil, nil
 	}
-	d := newDecbufAt(bs, off)
+	d := encoding.NewDecbufAt(bs, off, castagnoliTable)
 
 	var (
-		origLen     = d.len()
-		cnt         = d.be32int()
+		origLen     = d.Len()
+		cnt         = d.Be32int()
 		basePos     = uint32(off) + 4
-		nextPos     = basePos + uint32(origLen-d.len())
+		nextPos     = basePos + uint32(origLen-d.Len())
 		symbolSlice []string
 		symbols     = map[uint32]string{}
 	)
@@ -758,35 +753,35 @@ func ReadSymbols(bs ByteSlice, version int, off int) ([]string, map[uint32]strin
 		symbolSlice = make([]string, 0, cnt)
 	}
 
-	for d.err() == nil && d.len() > 0 && cnt > 0 {
-		s := d.uvarintStr()
+	for d.Err() == nil && d.Len() > 0 && cnt > 0 {
+		s := d.UvarintStr()
 
 		if version == FormatV2 {
 			symbolSlice = append(symbolSlice, s)
 		} else {
 			symbols[nextPos] = s
-			nextPos = basePos + uint32(origLen-d.len())
+			nextPos = basePos + uint32(origLen-d.Len())
 		}
 		cnt--
 	}
-	return symbolSlice, symbols, errors.Wrap(d.err(), "read symbols")
+	return symbolSlice, symbols, errors.Wrap(d.Err(), "read symbols")
 }
 
 // ReadOffsetTable reads an offset table and at the given position calls f for each
 // found entry. If f returns an error it stops decoding and returns the received error.
 func ReadOffsetTable(bs ByteSlice, off uint64, f func([]string, uint64) error) error {
-	d := newDecbufAt(bs, int(off))
-	cnt := d.be32()
+	d := encoding.NewDecbufAt(bs, int(off), castagnoliTable)
+	cnt := d.Be32()
 
-	for d.err() == nil && d.len() > 0 && cnt > 0 {
-		keyCount := d.uvarint()
+	for d.Err() == nil && d.Len() > 0 && cnt > 0 {
+		keyCount := d.Uvarint()
 		keys := make([]string, 0, keyCount)
 
 		for i := 0; i < keyCount; i++ {
-			keys = append(keys, d.uvarintStr())
+			keys = append(keys, d.UvarintStr())
 		}
-		o := d.uvarint64()
-		if d.err() != nil {
+		o := d.Uvarint64()
+		if d.Err() != nil {
 			break
 		}
 		if err := f(keys, o); err != nil {
@@ -794,7 +789,7 @@ func ReadOffsetTable(bs ByteSlice, off uint64, f func([]string, uint64) error) e
 		}
 		cnt--
 	}
-	return d.err()
+	return d.Err()
 }
 
 // Close the reader and its underlying resources.
@@ -843,17 +838,17 @@ func (r *Reader) LabelValues(names ...string) (StringTuples, error) {
 		//return nil, fmt.Errorf("label index doesn't exist")
 	}
 
-	d := newDecbufAt(r.b, int(off))
+	d := encoding.NewDecbufAt(r.b, int(off), castagnoliTable)
 
-	nc := d.be32int()
-	d.be32() // consume unused value entry count.
+	nc := d.Be32int()
+	d.Be32() // consume unused value entry count.
 
-	if d.err() != nil {
-		return nil, errors.Wrap(d.err(), "read label value index")
+	if d.Err() != nil {
+		return nil, errors.Wrap(d.Err(), "read label value index")
 	}
 	st := &serializedStringTuples{
 		idsCount: nc,
-		idsBytes: d.get(),
+		idsBytes: d.Get(),
 		lookup:   r.lookupSymbol,
 	}
 	return st, nil
@@ -882,11 +877,11 @@ func (r *Reader) Series(id uint64, lbls *labels.Labels, chks *[]chunks.Meta) err
 	if r.version == FormatV2 {
 		offset = id * 16
 	}
-	d := newDecbufUvarintAt(r.b, int(offset))
-	if d.err() != nil {
-		return d.err()
+	d := encoding.NewDecbufUvarintAt(r.b, int(offset), castagnoliTable)
+	if d.Err() != nil {
+		return d.Err()
 	}
-	return errors.Wrap(r.dec.Series(d.get(), lbls, chks), "read series")
+	return errors.Wrap(r.dec.Series(d.Get(), lbls, chks), "read series")
 }
 
 // Postings returns a postings list for the given label pair.
@@ -899,11 +894,11 @@ func (r *Reader) Postings(name, value string) (Postings, error) {
 	if !ok {
 		return EmptyPostings(), nil
 	}
-	d := newDecbufAt(r.b, int(off))
-	if d.err() != nil {
-		return nil, errors.Wrap(d.err(), "get postings entry")
+	d := encoding.NewDecbufAt(r.b, int(off), castagnoliTable)
+	if d.Err() != nil {
+		return nil, errors.Wrap(d.Err(), "get postings entry")
 	}
-	_, p, err := r.dec.Postings(d.get())
+	_, p, err := r.dec.Postings(d.Get())
 	if err != nil {
 		return nil, errors.Wrap(err, "decode postings")
 	}
@@ -952,7 +947,7 @@ type stringTuples struct {
 
 func NewStringTuples(entries []string, length int) (*stringTuples, error) {
 	if len(entries)%length != 0 {
-		return nil, errors.Wrap(errInvalidSize, "string tuple list")
+		return nil, errors.Wrap(encoding.ErrInvalidSize, "string tuple list")
 	}
 	return &stringTuples{entries: entries, length: length}, nil
 }
@@ -996,7 +991,7 @@ func (t *serializedStringTuples) Len() int {
 
 func (t *serializedStringTuples) At(i int) ([]string, error) {
 	if len(t.idsBytes) < (i+t.idsCount)*4 {
-		return nil, errInvalidSize
+		return nil, encoding.ErrInvalidSize
 	}
 	res := make([]string, 0, t.idsCount)
 
@@ -1023,10 +1018,10 @@ type Decoder struct {
 
 // Postings returns a postings list for b and its number of elements.
 func (dec *Decoder) Postings(b []byte) (int, Postings, error) {
-	d := decbuf{b: b}
-	n := d.be32int()
-	l := d.get()
-	return n, newBigEndianPostings(l), d.err()
+	d := encoding.Decbuf{B: b}
+	n := d.Be32int()
+	l := d.Get()
+	return n, newBigEndianPostings(l), d.Err()
 }
 
 // Series decodes a series entry from the given byte slice into lset and chks.
@@ -1034,16 +1029,16 @@ func (dec *Decoder) Series(b []byte, lbls *labels.Labels, chks *[]chunks.Meta) e
 	*lbls = (*lbls)[:0]
 	*chks = (*chks)[:0]
 
-	d := decbuf{b: b}
+	d := encoding.Decbuf{B: b}
 
-	k := d.uvarint()
+	k := d.Uvarint()
 
 	for i := 0; i < k; i++ {
-		lno := uint32(d.uvarint())
-		lvo := uint32(d.uvarint())
+		lno := uint32(d.Uvarint())
+		lvo := uint32(d.Uvarint())
 
-		if d.err() != nil {
-			return errors.Wrap(d.err(), "read series label offsets")
+		if d.Err() != nil {
+			return errors.Wrap(d.Err(), "read series label offsets")
 		}
 
 		ln, err := dec.LookupSymbol(lno)
@@ -1059,15 +1054,15 @@ func (dec *Decoder) Series(b []byte, lbls *labels.Labels, chks *[]chunks.Meta) e
 	}
 
 	// Read the chunks meta data.
-	k = d.uvarint()
+	k = d.Uvarint()
 
 	if k == 0 {
 		return nil
 	}
 
-	t0 := d.varint64()
-	maxt := int64(d.uvarint64()) + t0
-	ref0 := int64(d.uvarint64())
+	t0 := d.Varint64()
+	maxt := int64(d.Uvarint64()) + t0
+	ref0 := int64(d.Uvarint64())
 
 	*chks = append(*chks, chunks.Meta{
 		Ref:     uint64(ref0),
@@ -1077,14 +1072,14 @@ func (dec *Decoder) Series(b []byte, lbls *labels.Labels, chks *[]chunks.Meta) e
 	t0 = maxt
 
 	for i := 1; i < k; i++ {
-		mint := int64(d.uvarint64()) + t0
-		maxt := int64(d.uvarint64()) + mint
+		mint := int64(d.Uvarint64()) + t0
+		maxt := int64(d.Uvarint64()) + mint
 
-		ref0 += d.varint64()
+		ref0 += d.Varint64()
 		t0 = maxt
 
-		if d.err() != nil {
-			return errors.Wrapf(d.err(), "read meta for chunk %d", i)
+		if d.Err() != nil {
+			return errors.Wrapf(d.Err(), "read meta for chunk %d", i)
 		}
 
 		*chks = append(*chks, chunks.Meta{
@@ -1093,5 +1088,5 @@ func (dec *Decoder) Series(b []byte, lbls *labels.Labels, chks *[]chunks.Meta) e
 			MaxTime: maxt,
 		})
 	}
-	return d.err()
+	return d.Err()
 }
diff --git a/record.go b/record.go
index 364e814..8d9c575 100644
--- a/record.go
+++ b/record.go
@@ -19,6 +19,7 @@ import (
 	"sort"
 
 	"github.com/pkg/errors"
+	"github.com/prometheus/tsdb/encoding"
 	"github.com/prometheus/tsdb/labels"
 )
 
@@ -56,19 +57,19 @@ func (d *RecordDecoder) Type(rec []byte) RecordType {
 
 // Series appends series in rec to the given slice.
 func (d *RecordDecoder) Series(rec []byte, series []RefSeries) ([]RefSeries, error) {
-	dec := decbuf{b: rec}
+	dec := encoding.Decbuf{B: rec}
 
-	if RecordType(dec.byte()) != RecordSeries {
+	if RecordType(dec.Byte()) != RecordSeries {
 		return nil, errors.New("invalid record type")
 	}
-	for len(dec.b) > 0 && dec.err() == nil {
-		ref := dec.be64()
+	for len(dec.B) > 0 && dec.Err() == nil {
+		ref := dec.Be64()
 
-		lset := make(labels.Labels, dec.uvarint())
+		lset := make(labels.Labels, dec.Uvarint())
 
 		for i := range lset {
-			lset[i].Name = dec.uvarintStr()
-			lset[i].Value = dec.uvarintStr()
+			lset[i].Name = dec.UvarintStr()
+			lset[i].Value = dec.UvarintStr()
 		}
 		sort.Sort(lset)
 
@@ -77,33 +78,33 @@ func (d *RecordDecoder) Series(rec []byte, series []RefSeries) ([]RefSeries, err
 			Labels: lset,
 		})
 	}
-	if dec.err() != nil {
-		return nil, dec.err()
+	if dec.Err() != nil {
+		return nil, dec.Err()
 	}
-	if len(dec.b) > 0 {
-		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
+	if len(dec.B) > 0 {
+		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.B))
 	}
 	return series, nil
 }
 
 // Samples appends samples in rec to the given slice.
 func (d *RecordDecoder) Samples(rec []byte, samples []RefSample) ([]RefSample, error) {
-	dec := decbuf{b: rec}
+	dec := encoding.Decbuf{B: rec}
 
-	if RecordType(dec.byte()) != RecordSamples {
+	if RecordType(dec.Byte()) != RecordSamples {
 		return nil, errors.New("invalid record type")
 	}
-	if dec.len() == 0 {
+	if dec.Len() == 0 {
 		return samples, nil
 	}
 	var (
-		baseRef  = dec.be64()
-		baseTime = dec.be64int64()
+		baseRef  = dec.Be64()
+		baseTime = dec.Be64int64()
 	)
-	for len(dec.b) > 0 && dec.err() == nil {
-		dref := dec.varint64()
-		dtime := dec.varint64()
-		val := dec.be64()
+	for len(dec.B) > 0 && dec.Err() == nil {
+		dref := dec.Varint64()
+		dtime := dec.Varint64()
+		val := dec.Be64()
 
 		samples = append(samples, RefSample{
 			Ref: uint64(int64(baseRef) + dref),
@@ -112,35 +113,35 @@ func (d *RecordDecoder) Samples(rec []byte, samples []RefSample) ([]RefSample, e
 		})
 	}
 
-	if dec.err() != nil {
-		return nil, errors.Wrapf(dec.err(), "decode error after %d samples", len(samples))
+	if dec.Err() != nil {
+		return nil, errors.Wrapf(dec.Err(), "decode error after %d samples", len(samples))
 	}
-	if len(dec.b) > 0 {
-		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
+	if len(dec.B) > 0 {
+		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.B))
 	}
 	return samples, nil
 }
 
 // Tombstones appends tombstones in rec to the given slice.
 func (d *RecordDecoder) Tombstones(rec []byte, tstones []Stone) ([]Stone, error) {
-	dec := decbuf{b: rec}
+	dec := encoding.Decbuf{B: rec}
 
-	if RecordType(dec.byte()) != RecordTombstones {
+	if RecordType(dec.Byte()) != RecordTombstones {
 		return nil, errors.New("invalid record type")
 	}
-	for dec.len() > 0 && dec.err() == nil {
+	for dec.Len() > 0 && dec.Err() == nil {
 		tstones = append(tstones, Stone{
-			ref: dec.be64(),
+			ref: dec.Be64(),
 			intervals: Intervals{
-				{Mint: dec.varint64(), Maxt: dec.varint64()},
+				{Mint: dec.Varint64(), Maxt: dec.Varint64()},
 			},
 		})
 	}
-	if dec.err() != nil {
-		return nil, dec.err()
+	if dec.Err() != nil {
+		return nil, dec.Err()
 	}
-	if len(dec.b) > 0 {
-		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
+	if len(dec.B) > 0 {
+		return nil, errors.Errorf("unexpected %d bytes left in entry", len(dec.B))
 	}
 	return tstones, nil
 }
@@ -152,56 +153,56 @@ type RecordEncoder struct {
 
 // Series appends the encoded series to b and returns the resulting slice.
 func (e *RecordEncoder) Series(series []RefSeries, b []byte) []byte {
-	buf := encbuf{b: b}
-	buf.putByte(byte(RecordSeries))
+	buf := encoding.Encbuf{B: b}
+	buf.PutByte(byte(RecordSeries))
 
 	for _, s := range series {
-		buf.putBE64(s.Ref)
-		buf.putUvarint(len(s.Labels))
+		buf.PutBE64(s.Ref)
+		buf.PutUvarint(len(s.Labels))
 
 		for _, l := range s.Labels {
-			buf.putUvarintStr(l.Name)
-			buf.putUvarintStr(l.Value)
+			buf.PutUvarintStr(l.Name)
+			buf.PutUvarintStr(l.Value)
 		}
 	}
-	return buf.get()
+	return buf.Get()
 }
 
 // Samples appends the encoded samples to b and returns the resulting slice.
 func (e *RecordEncoder) Samples(samples []RefSample, b []byte) []byte {
-	buf := encbuf{b: b}
-	buf.putByte(byte(RecordSamples))
+	buf := encoding.Encbuf{B: b}
+	buf.PutByte(byte(RecordSamples))
 
 	if len(samples) == 0 {
-		return buf.get()
+		return buf.Get()
 	}
 
 	// Store base timestamp and base reference number of first sample.
 	// All samples encode their timestamp and ref as delta to those.
 	first := samples[0]
 
-	buf.putBE64(first.Ref)
-	buf.putBE64int64(first.T)
+	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))
+		buf.PutVarint64(int64(s.Ref) - int64(first.Ref))
+		buf.PutVarint64(s.T - first.T)
+		buf.PutBE64(math.Float64bits(s.V))
 	}
-	return buf.get()
+	return buf.Get()
 }
 
 // Tombstones appends the encoded tombstones to b and returns the resulting slice.
 func (e *RecordEncoder) Tombstones(tstones []Stone, b []byte) []byte {
-	buf := encbuf{b: b}
-	buf.putByte(byte(RecordTombstones))
+	buf := encoding.Encbuf{B: b}
+	buf.PutByte(byte(RecordTombstones))
 
 	for _, s := range tstones {
 		for _, iv := range s.intervals {
-			buf.putBE64(s.ref)
-			buf.putVarint64(iv.Mint)
-			buf.putVarint64(iv.Maxt)
+			buf.PutBE64(s.ref)
+			buf.PutVarint64(iv.Mint)
+			buf.PutVarint64(iv.Maxt)
 		}
 	}
-	return buf.get()
+	return buf.Get()
 }
diff --git a/tombstones.go b/tombstones.go
index 0781404..4c36c7b 100644
--- a/tombstones.go
+++ b/tombstones.go
@@ -23,6 +23,7 @@ import (
 	"sync"
 
 	"github.com/pkg/errors"
+	"github.com/prometheus/tsdb/encoding"
 )
 
 const tombstoneFilename = "tombstones"
@@ -64,12 +65,12 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
 		}
 	}()
 
-	buf := encbuf{b: make([]byte, 3*binary.MaxVarintLen64)}
-	buf.reset()
+	buf := encoding.Encbuf{B: make([]byte, 3*binary.MaxVarintLen64)}
+	buf.Reset()
 	// Write the meta.
-	buf.putBE32(MagicTombstone)
-	buf.putByte(tombstoneFormatV1)
-	_, err = f.Write(buf.get())
+	buf.PutBE32(MagicTombstone)
+	buf.PutByte(tombstoneFormatV1)
+	_, err = f.Write(buf.Get())
 	if err != nil {
 		return err
 	}
@@ -78,13 +79,13 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
 
 	if err := tr.Iter(func(ref uint64, ivs Intervals) error {
 		for _, iv := range ivs {
-			buf.reset()
+			buf.Reset()
 
-			buf.putUvarint64(ref)
-			buf.putVarint64(iv.Mint)
-			buf.putVarint64(iv.Maxt)
+			buf.PutUvarint64(ref)
+			buf.PutVarint64(iv.Mint)
+			buf.PutVarint64(iv.Maxt)
 
-			_, err = mw.Write(buf.get())
+			_, err = mw.Write(buf.Get())
 			if err != nil {
 				return err
 			}
@@ -126,24 +127,24 @@ func readTombstones(dir string) (TombstoneReader, SizeReader, error) {
 	}
 
 	if len(b) < 5 {
-		return nil, sr, errors.Wrap(errInvalidSize, "tombstones header")
+		return nil, sr, errors.Wrap(encoding.ErrInvalidSize, "tombstones header")
 	}
 
-	d := &decbuf{b: b[:len(b)-4]} // 4 for the checksum.
-	if mg := d.be32(); mg != MagicTombstone {
+	d := &encoding.Decbuf{B: b[:len(b)-4]} // 4 for the checksum.
+	if mg := d.Be32(); mg != MagicTombstone {
 		return nil, sr, fmt.Errorf("invalid magic number %x", mg)
 	}
-	if flag := d.byte(); flag != tombstoneFormatV1 {
+	if flag := d.Byte(); flag != tombstoneFormatV1 {
 		return nil, sr, fmt.Errorf("invalid tombstone format %x", flag)
 	}
 
-	if d.err() != nil {
-		return nil, sr, d.err()
+	if d.Err() != nil {
+		return nil, sr, d.Err()
 	}
 
 	// Verify checksum.
 	hash := newCRC32()
-	if _, err := hash.Write(d.get()); err != nil {
+	if _, err := hash.Write(d.Get()); err != nil {
 		return nil, sr, errors.Wrap(err, "write to hash")
 	}
 	if binary.BigEndian.Uint32(b[len(b)-4:]) != hash.Sum32() {
@@ -152,12 +153,12 @@ func readTombstones(dir string) (TombstoneReader, SizeReader, error) {
 
 	stonesMap := newMemTombstones()
 
-	for d.len() > 0 {
-		k := d.uvarint64()
-		mint := d.varint64()
-		maxt := d.varint64()
-		if d.err() != nil {
-			return nil, sr, d.err()
+	for d.Len() > 0 {
+		k := d.Uvarint64()
+		mint := d.Varint64()
+		maxt := d.Varint64()
+		if d.Err() != nil {
+			return nil, sr, d.Err()
 		}
 
 		stonesMap.addInterval(k, Interval{mint, maxt})
diff --git a/wal.go b/wal.go
index 60e1c58..d7ffe0c 100644
--- a/wal.go
+++ b/wal.go
@@ -31,6 +31,7 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/tsdb/encoding"
 	"github.com/prometheus/tsdb/fileutil"
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/wal"
@@ -287,16 +288,16 @@ func (w *SegmentWAL) Reader() WALReader {
 	}
 }
 
-func (w *SegmentWAL) getBuffer() *encbuf {
+func (w *SegmentWAL) getBuffer() *encoding.Encbuf {
 	b := w.buffers.Get()
 	if b == nil {
-		return &encbuf{b: make([]byte, 0, 64*1024)}
+		return &encoding.Encbuf{B: make([]byte, 0, 64*1024)}
 	}
-	return b.(*encbuf)
+	return b.(*encoding.Encbuf)
 }
 
-func (w *SegmentWAL) putBuffer(b *encbuf) {
-	b.reset()
+func (w *SegmentWAL) putBuffer(b *encoding.Encbuf) {
+	b.Reset()
 	w.buffers.Put(b)
 }
 
@@ -366,7 +367,7 @@ func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
 		buf := w.getBuffer()
 		flag = w.encodeSeries(buf, activeSeries)
 
-		_, err = w.writeTo(csf, crc32, WALEntrySeries, flag, buf.get())
+		_, err = w.writeTo(csf, crc32, WALEntrySeries, flag, buf.Get())
 		w.putBuffer(buf)
 
 		if err != nil {
@@ -427,7 +428,7 @@ func (w *SegmentWAL) LogSeries(series []RefSeries) error {
 	w.mtx.Lock()
 	defer w.mtx.Unlock()
 
-	err := w.write(WALEntrySeries, flag, buf.get())
+	err := w.write(WALEntrySeries, flag, buf.Get())
 
 	w.putBuffer(buf)
 
@@ -454,7 +455,7 @@ func (w *SegmentWAL) LogSamples(samples []RefSample) error {
 	w.mtx.Lock()
 	defer w.mtx.Unlock()
 
-	err := w.write(WALEntrySamples, flag, buf.get())
+	err := w.write(WALEntrySamples, flag, buf.Get())
 
 	w.putBuffer(buf)
 
@@ -480,7 +481,7 @@ func (w *SegmentWAL) LogDeletes(stones []Stone) error {
 	w.mtx.Lock()
 	defer w.mtx.Unlock()
 
-	err := w.write(WALEntryDeletes, flag, buf.get())
+	err := w.write(WALEntryDeletes, flag, buf.Get())
 
 	w.putBuffer(buf)
 
@@ -783,20 +784,20 @@ const (
 	walDeletesSimple = 1
 )
 
-func (w *SegmentWAL) encodeSeries(buf *encbuf, series []RefSeries) uint8 {
+func (w *SegmentWAL) encodeSeries(buf *encoding.Encbuf, series []RefSeries) uint8 {
 	for _, s := range series {
-		buf.putBE64(s.Ref)
-		buf.putUvarint(len(s.Labels))
+		buf.PutBE64(s.Ref)
+		buf.PutUvarint(len(s.Labels))
 
 		for _, l := range s.Labels {
-			buf.putUvarintStr(l.Name)
-			buf.putUvarintStr(l.Value)
+			buf.PutUvarintStr(l.Name)
+			buf.PutUvarintStr(l.Value)
 		}
 	}
 	return walSeriesSimple
 }
 
-func (w *SegmentWAL) encodeSamples(buf *encbuf, samples []RefSample) uint8 {
+func (w *SegmentWAL) encodeSamples(buf *encoding.Encbuf, samples []RefSample) uint8 {
 	if len(samples) == 0 {
 		return walSamplesSimple
 	}
@@ -806,23 +807,23 @@ func (w *SegmentWAL) encodeSamples(buf *encbuf, samples []RefSample) uint8 {
 	// TODO(fabxc): optimize for all samples having the same timestamp.
 	first := samples[0]
 
-	buf.putBE64(first.Ref)
-	buf.putBE64int64(first.T)
+	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))
+		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 {
+func (w *SegmentWAL) encodeDeletes(buf *encoding.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)
+			buf.PutBE64(s.ref)
+			buf.PutVarint64(iv.Mint)
+			buf.PutVarint64(iv.Maxt)
 		}
 	}
 	return walDeletesSimple
@@ -1115,16 +1116,16 @@ func (r *walReader) entry(cr io.Reader) (WALEntryType, byte, []byte, error) {
 }
 
 func (r *walReader) decodeSeries(flag byte, b []byte, res *[]RefSeries) error {
-	dec := decbuf{b: b}
+	dec := encoding.Decbuf{B: b}
 
-	for len(dec.b) > 0 && dec.err() == nil {
-		ref := dec.be64()
+	for len(dec.B) > 0 && dec.Err() == nil {
+		ref := dec.Be64()
 
-		lset := make(labels.Labels, dec.uvarint())
+		lset := make(labels.Labels, dec.Uvarint())
 
 		for i := range lset {
-			lset[i].Name = dec.uvarintStr()
-			lset[i].Value = dec.uvarintStr()
+			lset[i].Name = dec.UvarintStr()
+			lset[i].Value = dec.UvarintStr()
 		}
 		sort.Sort(lset)
 
@@ -1133,11 +1134,11 @@ func (r *walReader) decodeSeries(flag byte, b []byte, res *[]RefSeries) error {
 			Labels: lset,
 		})
 	}
-	if dec.err() != nil {
-		return dec.err()
+	if dec.Err() != nil {
+		return dec.Err()
 	}
-	if len(dec.b) > 0 {
-		return errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
+	if len(dec.B) > 0 {
+		return errors.Errorf("unexpected %d bytes left in entry", len(dec.B))
 	}
 	return nil
 }
@@ -1146,17 +1147,17 @@ func (r *walReader) decodeSamples(flag byte, b []byte, res *[]RefSample) error {
 	if len(b) == 0 {
 		return nil
 	}
-	dec := decbuf{b: b}
+	dec := encoding.Decbuf{B: b}
 
 	var (
-		baseRef  = dec.be64()
-		baseTime = dec.be64int64()
+		baseRef  = dec.Be64()
+		baseTime = dec.Be64int64()
 	)
 
-	for len(dec.b) > 0 && dec.err() == nil {
-		dref := dec.varint64()
-		dtime := dec.varint64()
-		val := dec.be64()
+	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),
@@ -1165,31 +1166,31 @@ func (r *walReader) decodeSamples(flag byte, b []byte, res *[]RefSample) error {
 		})
 	}
 
-	if dec.err() != nil {
-		return errors.Wrapf(dec.err(), "decode error after %d samples", len(*res))
+	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))
+	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}
+	dec := &encoding.Decbuf{B: b}
 
-	for dec.len() > 0 && dec.err() == nil {
+	for dec.Len() > 0 && dec.Err() == nil {
 		*res = append(*res, Stone{
-			ref: dec.be64(),
+			ref: dec.Be64(),
 			intervals: Intervals{
-				{Mint: dec.varint64(), Maxt: dec.varint64()},
+				{Mint: dec.Varint64(), Maxt: dec.Varint64()},
 			},
 		})
 	}
-	if dec.err() != nil {
-		return dec.err()
+	if dec.Err() != nil {
+		return dec.Err()
 	}
-	if len(dec.b) > 0 {
-		return errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
+	if len(dec.B) > 0 {
+		return errors.Errorf("unexpected %d bytes left in entry", len(dec.B))
 	}
 	return nil
 }
diff --git a/wal/live_reader.go b/wal/live_reader.go
new file mode 100644
index 0000000..8394bfd
--- /dev/null
+++ b/wal/live_reader.go
@@ -0,0 +1,284 @@
+// Copyright 2019 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 wal
+
+import (
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"io"
+
+	"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/client_golang/prometheus/promauto"
+)
+
+var (
+	readerCorruptionErrors = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "prometheus_tsdb_wal_reader_corruption_errors",
+		Help: "Errors encountered when reading the WAL.",
+	}, []string{"error"})
+)
+
+// NewLiveReader returns a new live reader.
+func NewLiveReader(logger log.Logger, r io.Reader) *LiveReader {
+	return &LiveReader{
+		logger: logger,
+		rdr:    r,
+
+		// Until we understand how they come about, make readers permissive
+		// to records spanning pages.
+		permissive: true,
+	}
+}
+
+// LiveReader reads WAL records from an io.Reader. It allows reading of WALs
+// that are still in the process of being written, and returns records as soon
+// as they can be read.
+type LiveReader struct {
+	logger     log.Logger
+	rdr        io.Reader
+	err        error
+	rec        []byte
+	hdr        [recordHeaderSize]byte
+	buf        [pageSize]byte
+	readIndex  int   // Index in buf to start at for next read.
+	writeIndex int   // Index in buf to start at for next write.
+	total      int64 // Total bytes processed during reading in calls to Next().
+	index      int   // Used to track partial records, should be 0 at the start of every new record.
+
+	// For testing, we can treat EOF as a non-error.
+	eofNonErr bool
+
+	// We sometime see records span page boundaries.  Should never happen, but it
+	// does.  Until we track down why, set permissive to true to tolerate it.
+	// NB the non-ive Reader implementation allows for this.
+	permissive bool
+}
+
+// Err returns any errors encountered reading the WAL.  io.EOFs are not terminal
+// and Next can be tried again.  Non-EOFs are terminal, and the reader should
+// not be used again.  It is up to the user to decide when to stop trying should
+// io.EOF be returned.
+func (r *LiveReader) Err() error {
+	if r.eofNonErr && r.err == io.EOF {
+		return nil
+	}
+	return r.err
+}
+
+// Offset returns the number of bytes consumed from this segment.
+func (r *LiveReader) Offset() int64 {
+	return r.total
+}
+
+func (r *LiveReader) fillBuffer() (int, error) {
+	n, err := r.rdr.Read(r.buf[r.writeIndex:len(r.buf)])
+	r.writeIndex += n
+	return n, err
+}
+
+// Next returns true if Record() will contain a full record.
+// If Next returns false, you should always checked the contents of Error().
+// Return false guarantees there are no more records if the segment is closed
+// and not corrupt, otherwise if Err() == io.EOF you should try again when more
+// data has been written.
+func (r *LiveReader) Next() bool {
+	for {
+		// If buildRecord returns a non-EOF error, its game up - the segment is
+		// corrupt. If buildRecord returns an EOF, we try and read more in
+		// fillBuffer later on. If that fails to read anything (n=0 && err=EOF),
+		// we return  EOF and the user can try again later. If we have a full
+		// page, buildRecord is guaranteed to return a record or a non-EOF; it
+		// has checks the records fit in pages.
+		if ok, err := r.buildRecord(); ok {
+			return true
+		} else if err != nil && err != io.EOF {
+			r.err = err
+			return false
+		}
+
+		// If we've filled the page and not found a record, this
+		// means records have started to span pages.  Shouldn't happen
+		// but does and until we found out why, we need to deal with this.
+		if r.permissive && r.writeIndex == pageSize && r.readIndex > 0 {
+			copy(r.buf[:], r.buf[r.readIndex:])
+			r.writeIndex -= r.readIndex
+			r.readIndex = 0
+			continue
+		}
+
+		if r.readIndex == pageSize {
+			r.writeIndex = 0
+			r.readIndex = 0
+		}
+
+		if r.writeIndex != pageSize {
+			n, err := r.fillBuffer()
+			if n == 0 || (err != nil && err != io.EOF) {
+				r.err = err
+				return false
+			}
+		}
+	}
+}
+
+// Record returns the current record.
+// The returned byte slice is only valid until the next call to Next.
+func (r *LiveReader) Record() []byte {
+	return r.rec
+}
+
+// Rebuild a full record from potentially partial records. Returns false
+// if there was an error or if we weren't able to read a record for any reason.
+// Returns true if we read a full record. Any record data is appended to
+// LiveReader.rec
+func (r *LiveReader) buildRecord() (bool, error) {
+	for {
+		// Check that we have data in the internal buffer to read.
+		if r.writeIndex <= r.readIndex {
+			return false, nil
+		}
+
+		// Attempt to read a record, partial or otherwise.
+		temp, n, err := r.readRecord()
+		if err != nil {
+			return false, err
+		}
+
+		r.readIndex += n
+		r.total += int64(n)
+		if temp == nil {
+			return false, nil
+		}
+
+		rt := recType(r.hdr[0])
+		if rt == recFirst || rt == recFull {
+			r.rec = r.rec[:0]
+		}
+		r.rec = append(r.rec, temp...)
+
+		if err := validateRecord(rt, r.index); err != nil {
+			r.index = 0
+			return false, err
+		}
+		if rt == recLast || rt == recFull {
+			r.index = 0
+			return true, nil
+		}
+		// Only increment i for non-zero records since we use it
+		// to determine valid content record sequences.
+		r.index++
+	}
+}
+
+// Returns an error if the recType and i indicate an invalid record sequence.
+// As an example, if i is > 0 because we've read some amount of a partial record
+// (recFirst, recMiddle, etc. but not recLast) and then we get another recFirst or recFull
+// instead of a recLast or recMiddle we would have an invalid record.
+func validateRecord(typ recType, i int) error {
+	switch typ {
+	case recFull:
+		if i != 0 {
+			return errors.New("unexpected full record")
+		}
+		return nil
+	case recFirst:
+		if i != 0 {
+			return errors.New("unexpected first record, dropping buffer")
+		}
+		return nil
+	case recMiddle:
+		if i == 0 {
+			return errors.New("unexpected middle record, dropping buffer")
+		}
+		return nil
+	case recLast:
+		if i == 0 {
+			return errors.New("unexpected last record, dropping buffer")
+		}
+		return nil
+	default:
+		return errors.Errorf("unexpected record type %d", typ)
+	}
+}
+
+// Read a sub-record (see recType) from the buffer. It could potentially
+// be a full record (recFull) if the record fits within the bounds of a single page.
+// Returns a byte slice of the record data read, the number of bytes read, and an error
+// if there's a non-zero byte in a page term record or the record checksum fails.
+// This is a non-method function to make it clear it does not mutate the reader.
+func (r *LiveReader) readRecord() ([]byte, int, error) {
+	// Special case: for recPageTerm, check that are all zeros to end of page,
+	// consume them but don't return them.
+	if r.buf[r.readIndex] == byte(recPageTerm) {
+		// End of page won't necessarily be end of buffer, as we may have
+		// got misaligned by records spanning page boundaries.
+		// r.total % pageSize is the offset into the current page
+		// that r.readIndex points to in buf.  Therefore
+		// pageSize - (r.total % pageSize) is the amount left to read of
+		// the current page.
+		remaining := int(pageSize - (r.total % pageSize))
+		if r.readIndex+remaining > r.writeIndex {
+			return nil, 0, io.EOF
+		}
+
+		for i := r.readIndex; i < r.readIndex+remaining; i++ {
+			if r.buf[i] != 0 {
+				return nil, 0, errors.New("unexpected non-zero byte in page term bytes")
+			}
+		}
+
+		return nil, remaining, nil
+	}
+
+	// Not a recPageTerm; read the record and check the checksum.
+	if r.writeIndex-r.readIndex < recordHeaderSize {
+		return nil, 0, io.EOF
+	}
+
+	copy(r.hdr[:], r.buf[r.readIndex:r.readIndex+recordHeaderSize])
+	length := int(binary.BigEndian.Uint16(r.hdr[1:]))
+	crc := binary.BigEndian.Uint32(r.hdr[3:])
+	if r.readIndex+recordHeaderSize+length > pageSize {
+		if !r.permissive {
+			return nil, 0, fmt.Errorf("record would overflow current page: %d > %d", r.readIndex+recordHeaderSize+length, pageSize)
+		}
+		readerCorruptionErrors.WithLabelValues("record_span_page").Inc()
+		level.Warn(r.logger).Log("msg", "record spans page boundaries", "start", r.readIndex, "end", recordHeaderSize+length, "pageSize", pageSize)
+	}
+	if recordHeaderSize+length > pageSize {
+		return nil, 0, fmt.Errorf("record length greater than a single page: %d > %d", recordHeaderSize+length, pageSize)
+	}
+	if r.readIndex+recordHeaderSize+length > r.writeIndex {
+		return nil, 0, io.EOF
+	}
+
+	rec := r.buf[r.readIndex+recordHeaderSize : r.readIndex+recordHeaderSize+length]
+	if c := crc32.Checksum(rec, castagnoliTable); c != crc {
+		return nil, 0, errors.Errorf("unexpected checksum %x, expected %x", c, crc)
+	}
+
+	return rec, length + recordHeaderSize, nil
+}
+
+func min(i, j int) int {
+	if i < j {
+		return i
+	}
+	return j
+}
diff --git a/wal/reader.go b/wal/reader.go
new file mode 100644
index 0000000..297463b
--- /dev/null
+++ b/wal/reader.go
@@ -0,0 +1,183 @@
+// Copyright 2019 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 wal
+
+import (
+	"encoding/binary"
+	"hash/crc32"
+	"io"
+
+	"github.com/pkg/errors"
+)
+
+// Reader reads WAL records from an io.Reader.
+type Reader struct {
+	rdr       io.Reader
+	err       error
+	rec       []byte
+	buf       [pageSize]byte
+	total     int64   // Total bytes processed.
+	curRecTyp recType // Used for checking that the last record is not torn.
+}
+
+// NewReader returns a new reader.
+func NewReader(r io.Reader) *Reader {
+	return &Reader{rdr: r}
+}
+
+// Next advances the reader to the next records and returns true if it exists.
+// It must not be called again after it returned false.
+func (r *Reader) Next() bool {
+	err := r.next()
+	if errors.Cause(err) == io.EOF {
+		// The last WAL segment record shouldn't be torn(should be full or last).
+		// The last record would be torn after a crash just before
+		// the last record part could be persisted to disk.
+		if recType(r.curRecTyp) == recFirst || recType(r.curRecTyp) == recMiddle {
+			r.err = errors.New("last record is torn")
+		}
+		return false
+	}
+	r.err = err
+	return r.err == nil
+}
+
+func (r *Reader) next() (err error) {
+	// We have to use r.buf since allocating byte arrays here fails escape
+	// analysis and ends up on the heap, even though it seemingly should not.
+	hdr := r.buf[:recordHeaderSize]
+	buf := r.buf[recordHeaderSize:]
+
+	r.rec = r.rec[:0]
+
+	i := 0
+	for {
+		if _, err = io.ReadFull(r.rdr, hdr[:1]); err != nil {
+			return errors.Wrap(err, "read first header byte")
+		}
+		r.total++
+		r.curRecTyp = recType(hdr[0])
+
+		// Gobble up zero bytes.
+		if r.curRecTyp == recPageTerm {
+			// recPageTerm is a single byte that indicates the rest of the page is padded.
+			// If it's the first byte in a page, buf is too small and
+			// needs to be resized to fit pageSize-1 bytes.
+			buf = r.buf[1:]
+
+			// We are pedantic and check whether the zeros are actually up
+			// to a page boundary.
+			// It's not strictly necessary but may catch sketchy state early.
+			k := pageSize - (r.total % pageSize)
+			if k == pageSize {
+				continue // Initial 0 byte was last page byte.
+			}
+			n, err := io.ReadFull(r.rdr, buf[:k])
+			if err != nil {
+				return errors.Wrap(err, "read remaining zeros")
+			}
+			r.total += int64(n)
+
+			for _, c := range buf[:k] {
+				if c != 0 {
+					return errors.New("unexpected non-zero byte in padded page")
+				}
+			}
+			continue
+		}
+		n, err := io.ReadFull(r.rdr, hdr[1:])
+		if err != nil {
+			return errors.Wrap(err, "read remaining header")
+		}
+		r.total += int64(n)
+
+		var (
+			length = binary.BigEndian.Uint16(hdr[1:])
+			crc    = binary.BigEndian.Uint32(hdr[3:])
+		)
+
+		if length > pageSize-recordHeaderSize {
+			return errors.Errorf("invalid record size %d", length)
+		}
+		n, err = io.ReadFull(r.rdr, buf[:length])
+		if err != nil {
+			return err
+		}
+		r.total += int64(n)
+
+		if n != int(length) {
+			return errors.Errorf("invalid size: expected %d, got %d", length, n)
+		}
+		if c := crc32.Checksum(buf[:length], castagnoliTable); c != crc {
+			return errors.Errorf("unexpected checksum %x, expected %x", c, crc)
+		}
+		r.rec = append(r.rec, buf[:length]...)
+
+		if err := validateRecord(r.curRecTyp, i); err != nil {
+			return err
+		}
+		if r.curRecTyp == recLast || r.curRecTyp == recFull {
+			return nil
+		}
+
+		// Only increment i for non-zero records since we use it
+		// to determine valid content record sequences.
+		i++
+	}
+}
+
+// Err returns the last encountered error wrapped in a corruption error.
+// If the reader does not allow to infer a segment index and offset, a total
+// offset in the reader stream will be provided.
+func (r *Reader) Err() error {
+	if r.err == nil {
+		return nil
+	}
+	if b, ok := r.rdr.(*segmentBufReader); ok {
+		return &CorruptionErr{
+			Err:     r.err,
+			Dir:     b.segs[b.cur].Dir(),
+			Segment: b.segs[b.cur].Index(),
+			Offset:  int64(b.off),
+		}
+	}
+	return &CorruptionErr{
+		Err:     r.err,
+		Segment: -1,
+		Offset:  r.total,
+	}
+}
+
+// Record returns the current record. The returned byte slice is only
+// valid until the next call to Next.
+func (r *Reader) Record() []byte {
+	return r.rec
+}
+
+// Segment returns the current segment being read.
+func (r *Reader) Segment() int {
+	if b, ok := r.rdr.(*segmentBufReader); ok {
+		return b.segs[b.cur].Index()
+	}
+	return -1
+}
+
+// Offset returns the current position of the segment being read.
+func (r *Reader) Offset() int64 {
+	if b, ok := r.rdr.(*segmentBufReader); ok {
+		return int64(b.off)
+	}
+	return r.total
+}
diff --git a/wal/wal.go b/wal/wal.go
index 1ab6334..dd0be9c 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -681,20 +681,20 @@ func NewSegmentsRangeReader(sr ...SegmentRange) (io.ReadCloser, error) {
 
 // segmentBufReader is a buffered reader that reads in multiples of pages.
 // The main purpose is that we are able to track segment and offset for
-// corruption reporting.
+// corruption reporting.  We have to be careful not to increment curr too
+// early, as it is used by Reader.Err() to tell Repair which segment is corrupt.
+// As such we pad the end of non-page align segments with zeros.
 type segmentBufReader struct {
 	buf  *bufio.Reader
 	segs []*Segment
-	cur  int
-	off  int
-	more bool
+	cur  int // Index into segs.
+	off  int // Offset of read data into current segment.
 }
 
 func newSegmentBufReader(segs ...*Segment) *segmentBufReader {
 	return &segmentBufReader{
-		buf:  bufio.NewReaderSize(nil, 16*pageSize),
+		buf:  bufio.NewReaderSize(segs[0], 16*pageSize),
 		segs: segs,
-		cur:  -1,
 	}
 }
 
@@ -707,411 +707,38 @@ func (r *segmentBufReader) Close() (err error) {
 	return err
 }
 
+// Read implements io.Reader.
 func (r *segmentBufReader) Read(b []byte) (n int, err error) {
-	if !r.more {
-		if r.cur+1 >= len(r.segs) {
-			return 0, io.EOF
-		}
-		r.cur++
-		r.off = 0
-		r.more = true
-		r.buf.Reset(r.segs[r.cur])
-	}
 	n, err = r.buf.Read(b)
 	r.off += n
-	if err != io.EOF {
-		return n, err
-	}
-	// Just return what we read so far, but don't signal EOF.
-	// Only unset more so we don't invalidate the current segment and
-	// offset before the next read.
-	r.more = false
-	return n, nil
-}
-
-// Reader reads WAL records from an io.Reader.
-type Reader struct {
-	rdr       io.Reader
-	err       error
-	rec       []byte
-	buf       [pageSize]byte
-	total     int64   // Total bytes processed.
-	curRecTyp recType // Used for checking that the last record is not torn.
-}
-
-// NewReader returns a new reader.
-func NewReader(r io.Reader) *Reader {
-	return &Reader{rdr: r}
-}
-
-// Next advances the reader to the next records and returns true if it exists.
-// It must not be called again after it returned false.
-func (r *Reader) Next() bool {
-	err := r.next()
-	if errors.Cause(err) == io.EOF {
-		// The last WAL segment record shouldn't be torn(should be full or last).
-		// The last record would be torn after a crash just before
-		// the last record part could be persisted to disk.
-		if recType(r.curRecTyp) == recFirst || recType(r.curRecTyp) == recMiddle {
-			r.err = errors.New("last record is torn")
-		}
-		return false
-	}
-	r.err = err
-	return r.err == nil
-}
-
-func (r *Reader) next() (err error) {
-	// We have to use r.buf since allocating byte arrays here fails escape
-	// analysis and ends up on the heap, even though it seemingly should not.
-	hdr := r.buf[:recordHeaderSize]
-	buf := r.buf[recordHeaderSize:]
-
-	r.rec = r.rec[:0]
-
-	i := 0
-	for {
-		if _, err = io.ReadFull(r.rdr, hdr[:1]); err != nil {
-			return errors.Wrap(err, "read first header byte")
-		}
-		r.total++
-		r.curRecTyp = recType(hdr[0])
-
-		// Gobble up zero bytes.
-		if r.curRecTyp == recPageTerm {
-			// recPageTerm is a single byte that indicates the rest of the page is padded.
-			// If it's the first byte in a page, buf is too small and
-			// needs to be resized to fit pageSize-1 bytes.
-			buf = r.buf[1:]
-
-			// We are pedantic and check whether the zeros are actually up
-			// to a page boundary.
-			// It's not strictly necessary but may catch sketchy state early.
-			k := pageSize - (r.total % pageSize)
-			if k == pageSize {
-				continue // Initial 0 byte was last page byte.
-			}
-			n, err := io.ReadFull(r.rdr, buf[:k])
-			if err != nil {
-				return errors.Wrap(err, "read remaining zeros")
-			}
-			r.total += int64(n)
-
-			for _, c := range buf[:k] {
-				if c != 0 {
-					return errors.New("unexpected non-zero byte in padded page")
-				}
-			}
-			continue
-		}
-		n, err := io.ReadFull(r.rdr, hdr[1:])
-		if err != nil {
-			return errors.Wrap(err, "read remaining header")
-		}
-		r.total += int64(n)
-
-		var (
-			length = binary.BigEndian.Uint16(hdr[1:])
-			crc    = binary.BigEndian.Uint32(hdr[3:])
-		)
-
-		if length > pageSize-recordHeaderSize {
-			return errors.Errorf("invalid record size %d", length)
-		}
-		n, err = io.ReadFull(r.rdr, buf[:length])
-		if err != nil {
-			return err
-		}
-		r.total += int64(n)
-
-		if n != int(length) {
-			return errors.Errorf("invalid size: expected %d, got %d", length, n)
-		}
-		if c := crc32.Checksum(buf[:length], castagnoliTable); c != crc {
-			return errors.Errorf("unexpected checksum %x, expected %x", c, crc)
-		}
-		r.rec = append(r.rec, buf[:length]...)
-
-		if err := validateRecord(r.curRecTyp, i); err != nil {
-			return err
-		}
-		if r.curRecTyp == recLast || r.curRecTyp == recFull {
-			return nil
-		}
-
-		// Only increment i for non-zero records since we use it
-		// to determine valid content record sequences.
-		i++
-	}
-}
-
-// Err returns the last encountered error wrapped in a corruption error.
-// If the reader does not allow to infer a segment index and offset, a total
-// offset in the reader stream will be provided.
-func (r *Reader) Err() error {
-	if r.err == nil {
-		return nil
-	}
-	if b, ok := r.rdr.(*segmentBufReader); ok {
-		return &CorruptionErr{
-			Err:     r.err,
-			Dir:     b.segs[b.cur].Dir(),
-			Segment: b.segs[b.cur].Index(),
-			Offset:  int64(b.off),
-		}
-	}
-	return &CorruptionErr{
-		Err:     r.err,
-		Segment: -1,
-		Offset:  r.total,
-	}
-}
-
-// Record returns the current record. The returned byte slice is only
-// valid until the next call to Next.
-func (r *Reader) Record() []byte {
-	return r.rec
-}
-
-// Segment returns the current segment being read.
-func (r *Reader) Segment() int {
-	if b, ok := r.rdr.(*segmentBufReader); ok {
-		return b.segs[b.cur].Index()
-	}
-	return -1
-}
-
-// Offset returns the current position of the segment being read.
-func (r *Reader) Offset() int64 {
-	if b, ok := r.rdr.(*segmentBufReader); ok {
-		return int64(b.off)
-	}
-	return r.total
-}
-
-// NewLiveReader returns a new live reader.
-func NewLiveReader(r io.Reader) *LiveReader {
-	return &LiveReader{rdr: r}
-}
 
-// Reader reads WAL records from an io.Reader. It buffers partial record data for
-// the next read.
-type LiveReader struct {
-	rdr        io.Reader
-	err        error
-	rec        []byte
-	hdr        [recordHeaderSize]byte
-	buf        [pageSize]byte
-	readIndex  int   // Index in buf to start at for next read.
-	writeIndex int   // Index in buf to start at for next write.
-	total      int64 // Total bytes processed during reading in calls to Next().
-	index      int   // Used to track partial records, should be 0 at the start of every new record.
-}
-
-func (r *LiveReader) Err() error {
-	return r.err
-}
-
-func (r *LiveReader) TotalRead() int64 {
-	return r.total
-}
-
-func (r *LiveReader) fillBuffer() error {
-	n, err := r.rdr.Read(r.buf[r.writeIndex:len(r.buf)])
-	r.writeIndex += n
-	return err
-}
-
-// Shift the buffer up to the read index.
-func (r *LiveReader) shiftBuffer() {
-	copied := copy(r.buf[0:], r.buf[r.readIndex:r.writeIndex])
-	r.readIndex = 0
-	r.writeIndex = copied
-}
-
-// Next returns true if r.rec will contain a full record.
-// False does not indicate that there will never be more data to
-// read for the current io.Reader.
-func (r *LiveReader) Next() bool {
-	for {
-		if r.buildRecord() {
-			return true
-		}
-		if r.err != nil && r.err != io.EOF {
-			return false
-		}
-		if r.readIndex == pageSize {
-			r.shiftBuffer()
-		}
-		if r.writeIndex != pageSize {
-			if err := r.fillBuffer(); err != nil {
-				// We expect to get EOF, since we're reading the segment file as it's being written.
-				if err != io.EOF {
-					r.err = err
-				}
-				return false
-			}
-		}
-	}
-}
-
-// Record returns the current record.
-// The returned byte slice is only valid until the next call to Next.
-func (r *LiveReader) Record() []byte {
-	return r.rec
-}
-
-// Rebuild a full record from potentially partial records. Returns false
-// if there was an error or if we weren't able to read a record for any reason.
-// Returns true if we read a full record. Any record data is appeneded to
-// LiveReader.rec
-func (r *LiveReader) buildRecord() bool {
-	for {
-		// Check that we have data in the internal buffer to read.
-		if r.writeIndex <= r.readIndex {
-			return false
-		}
-
-		// Attempt to read a record, partial or otherwise.
-		temp, n, err := readRecord(r.buf[r.readIndex:r.writeIndex], r.hdr[:], r.total)
-		r.readIndex += n
-		r.total += int64(n)
-		if err != nil {
-			r.err = err
-			return false
-		}
-
-		if temp == nil {
-			return false
-		}
-
-		rt := recType(r.hdr[0])
-
-		if rt == recFirst || rt == recFull {
-			r.rec = r.rec[:0]
-		}
-		r.rec = append(r.rec, temp...)
-
-		if err := validateRecord(rt, r.index); err != nil {
-			r.err = err
-			r.index = 0
-			return false
-		}
-		if rt == recLast || rt == recFull {
-			r.index = 0
-			return true
-		}
-		// Only increment i for non-zero records since we use it
-		// to determine valid content record sequences.
-		r.index++
-	}
-}
-
-// Returns an error if the recType and i indicate an invalid record sequence.
-// As an example, if i is > 0 because we've read some amount of a partial record
-// (recFirst, recMiddle, etc. but not recLast) and then we get another recFirst or recFull
-// instead of a recLast or recMiddle we would have an invalid record.
-func validateRecord(typ recType, i int) error {
-	switch typ {
-	case recFull:
-		if i != 0 {
-			return errors.New("unexpected full record")
-		}
-		return nil
-	case recFirst:
-		if i != 0 {
-			return errors.New("unexpected first record, dropping buffer")
-		}
-		return nil
-	case recMiddle:
-		if i == 0 {
-			return errors.New("unexpected middle record, dropping buffer")
-		}
-		return nil
-	case recLast:
-		if i == 0 {
-			return errors.New("unexpected last record, dropping buffer")
-		}
-		return nil
-	default:
-		return errors.Errorf("unexpected record type %d", typ)
+	// If we succeeded, or hit a non-EOF, we can stop.
+	if err == nil || err != io.EOF {
+		return n, err
 	}
-}
 
-// Read a sub-record (see recType) from the buffer. It could potentially
-// be a full record (recFull) if the record fits within the bounds of a single page.
-// Returns a byte slice of the record data read, the number of bytes read, and an error
-// if there's a non-zero byte in a page term record or the record checksum fails.
-// TODO(callum) the EOF errors we're returning from this function should theoretically
-// never happen, add a metric for them.
-func readRecord(buf []byte, header []byte, total int64) ([]byte, int, error) {
-	readIndex := 0
-	header[0] = buf[0]
-	readIndex++
-	total++
-
-	// The rest of this function is mostly from Reader.Next().
-	typ := recType(header[0])
-	// Gobble up zero bytes.
-	if typ == recPageTerm {
-		// We are pedantic and check whether the zeros are actually up to a page boundary.
-		// It's not strictly necessary but may catch sketchy state early.
-		k := pageSize - (total % pageSize)
-		if k == pageSize {
-			return nil, 1, nil // Initial 0 byte was last page byte.
+	// We hit EOF; fake out zero padding at the end of short segments, so we
+	// don't increment curr too early and report the wrong segment as corrupt.
+	if r.off%pageSize != 0 {
+		i := 0
+		for ; n+i < len(b) && (r.off+i)%pageSize != 0; i++ {
+			b[n+i] = 0
 		}
 
-		if k <= int64(len(buf)-readIndex) {
-			for _, v := range buf[readIndex : int64(readIndex)+k] {
-				readIndex++
-				if v != 0 {
-					return nil, readIndex, errors.New("unexpected non-zero byte in page term bytes")
-				}
-			}
-			return nil, readIndex, nil
-		}
-		// Not enough bytes to read the rest of the page term rec.
-		// This theoretically should never happen, since we're only shifting the
-		// internal buffer of the live reader when we read to the end of page.
-		// Treat this the same as an EOF, it's an error we would expect to see.
-		return nil, 0, io.EOF
+		// Return early, even if we didn't fill b.
+		r.off += i
+		return n + i, nil
 	}
 
-	if readIndex+recordHeaderSize-1 > len(buf) {
-		// Treat this the same as an EOF, it's an error we would expect to see.
-		return nil, 0, io.EOF
+	// There is no more deta left in the curr segment and there are no more
+	// segments left.  Return EOF.
+	if r.cur+1 >= len(r.segs) {
+		return n, io.EOF
 	}
 
-	copy(header[1:], buf[readIndex:readIndex+len(header[1:])])
-	readIndex += recordHeaderSize - 1
-	total += int64(recordHeaderSize - 1)
-	var (
-		length = binary.BigEndian.Uint16(header[1:])
-		crc    = binary.BigEndian.Uint32(header[3:])
-	)
-	readTo := int(length) + readIndex
-	if readTo > len(buf) {
-		if (readTo - readIndex) > pageSize {
-			return nil, 0, errors.Errorf("invalid record, record size would be larger than max page size: %d", int(length))
-		}
-		// Not enough data to read all of the record data.
-		// Treat this the same as an EOF, it's an error we would expect to see.
-		return nil, 0, io.EOF
-	}
-	recData := buf[readIndex:readTo]
-	readIndex += int(length)
-	total += int64(length)
-
-	// TODO(callum) what should we do here, throw out the record? We should add a metric at least.
-	if c := crc32.Checksum(recData, castagnoliTable); c != crc {
-		return recData, readIndex, errors.Errorf("unexpected checksum %x, expected %x", c, crc)
-	}
-	return recData, readIndex, nil
-}
-
-func min(i, j int) int {
-	if i < j {
-		return i
-	}
-	return j
+	// Move to next segment.
+	r.cur++
+	r.off = 0
+	r.buf.Reset(r.segs[r.cur])
+	return n, nil
 }


Test patch:
diff --git a/index/index_test.go b/index/index_test.go
index 25c5be9..1395a34 100644
--- a/index/index_test.go
+++ b/index/index_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/tsdb/chunks"
+	"github.com/prometheus/tsdb/encoding"
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/testutil"
 )
@@ -395,8 +396,8 @@ func TestPersistence_index_e2e(t *testing.T) {
 func TestDecbufUvariantWithInvalidBuffer(t *testing.T) {
 	b := realByteSlice([]byte{0x81, 0x81, 0x81, 0x81, 0x81, 0x81})
 
-	db := newDecbufUvarintAt(b, 0)
-	testutil.NotOk(t, db.err())
+	db := encoding.NewDecbufUvarintAt(b, 0, castagnoliTable)
+	testutil.NotOk(t, db.Err())
 }
 
 func TestReaderWithInvalidBuffer(t *testing.T) {
diff --git a/testutil/logging.go b/testutil/logging.go
new file mode 100644
index 0000000..839b866
--- /dev/null
+++ b/testutil/logging.go
@@ -0,0 +1,35 @@
+// Copyright 2019 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 testutil
+
+import (
+	"testing"
+
+	"github.com/go-kit/kit/log"
+)
+
+type logger struct {
+	t *testing.T
+}
+
+// NewLogger returns a gokit compatible Logger which calls t.Log.
+func NewLogger(t *testing.T) log.Logger {
+	return logger{t: t}
+}
+
+// Log implements log.Logger.
+func (t logger) Log(keyvals ...interface{}) error {
+	t.t.Log(keyvals...)
+	return nil
+}
diff --git a/wal/reader_test.go b/wal/reader_test.go
new file mode 100644
index 0000000..d782ae8
--- /dev/null
+++ b/wal/reader_test.go
@@ -0,0 +1,506 @@
+// Copyright 2019 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 wal
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/prometheus/tsdb/testutil"
+)
+
+type reader interface {
+	Next() bool
+	Err() error
+	Record() []byte
+	Offset() int64
+}
+
+type record struct {
+	t recType
+	b []byte
+}
+
+var readerConstructors = map[string]func(io.Reader) reader{
+	"Reader": func(r io.Reader) reader {
+		return NewReader(r)
+	},
+	"LiveReader": func(r io.Reader) reader {
+		lr := NewLiveReader(log.NewNopLogger(), r)
+		lr.eofNonErr = true
+		return lr
+	},
+}
+
+var data = make([]byte, 100000)
+var testReaderCases = []struct {
+	t    []record
+	exp  [][]byte
+	fail bool
+}{
+	// Sequence of valid records.
+	{
+		t: []record{
+			{recFull, data[0:200]},
+			{recFirst, data[200:300]},
+			{recLast, data[300:400]},
+			{recFirst, data[400:800]},
+			{recMiddle, data[800:900]},
+			{recPageTerm, make([]byte, pageSize-900-recordHeaderSize*5-1)}, // exactly lines up with page boundary.
+			{recLast, data[900:900]},
+			{recFirst, data[900:1000]},
+			{recMiddle, data[1000:1200]},
+			{recMiddle, data[1200:30000]},
+			{recMiddle, data[30000:30001]},
+			{recMiddle, data[30001:30001]},
+			{recLast, data[30001:32000]},
+		},
+		exp: [][]byte{
+			data[0:200],
+			data[200:400],
+			data[400:900],
+			data[900:32000],
+		},
+	},
+	// Exactly at the limit of one page minus the header size
+	{
+		t: []record{
+			{recFull, data[0 : pageSize-recordHeaderSize]},
+		},
+		exp: [][]byte{
+			data[:pageSize-recordHeaderSize],
+		},
+	},
+	// More than a full page, this exceeds our buffer and can never happen
+	// when written by the WAL.
+	{
+		t: []record{
+			{recFull, data[0 : pageSize+1]},
+		},
+		fail: true,
+	},
+	// Two records the together are too big for a page.
+	// NB currently the non-live reader succeeds on this. I think this is a bug.
+	// but we've seen it in production.
+	{
+		t: []record{
+			{recFull, data[:pageSize/2]},
+			{recFull, data[:pageSize/2]},
+		},
+		exp: [][]byte{
+			data[:pageSize/2],
+			data[:pageSize/2],
+		},
+	},
+	// Invalid orders of record types.
+	{
+		t:    []record{{recMiddle, data[:200]}},
+		fail: true,
+	},
+	{
+		t:    []record{{recLast, data[:200]}},
+		fail: true,
+	},
+	{
+		t: []record{
+			{recFirst, data[:200]},
+			{recFull, data[200:400]},
+		},
+		fail: true,
+	},
+	{
+		t: []record{
+			{recFirst, data[:100]},
+			{recMiddle, data[100:200]},
+			{recFull, data[200:400]},
+		},
+		fail: true,
+	},
+	// Non-zero data after page termination.
+	{
+		t: []record{
+			{recFull, data[:100]},
+			{recPageTerm, append(make([]byte, pageSize-recordHeaderSize-102), 1)},
+		},
+		exp:  [][]byte{data[:100]},
+		fail: true,
+	},
+}
+
+func encodedRecord(t recType, b []byte) []byte {
+	if t == recPageTerm {
+		return append([]byte{0}, b...)
+	}
+	r := make([]byte, recordHeaderSize)
+	r[0] = byte(t)
+	binary.BigEndian.PutUint16(r[1:], uint16(len(b)))
+	binary.BigEndian.PutUint32(r[3:], crc32.Checksum(b, castagnoliTable))
+	return append(r, b...)
+}
+
+// TestReader feeds the reader a stream of encoded records with different types.
+func TestReader(t *testing.T) {
+	for name, fn := range readerConstructors {
+		for i, c := range testReaderCases {
+			t.Run(fmt.Sprintf("%s/%d", name, i), func(t *testing.T) {
+				var buf []byte
+				for _, r := range c.t {
+					buf = append(buf, encodedRecord(r.t, r.b)...)
+				}
+				r := fn(bytes.NewReader(buf))
+
+				for j := 0; r.Next(); j++ {
+					t.Logf("record %d", j)
+					rec := r.Record()
+
+					if j >= len(c.exp) {
+						t.Fatal("received more records than expected")
+					}
+					testutil.Equals(t, c.exp[j], rec, "Bytes within record did not match expected Bytes")
+				}
+				if !c.fail && r.Err() != nil {
+					t.Fatalf("unexpected error: %s", r.Err())
+				}
+				if c.fail && r.Err() == nil {
+					t.Fatalf("expected error but got none")
+				}
+			})
+		}
+	}
+}
+
+func TestReader_Live(t *testing.T) {
+	logger := testutil.NewLogger(t)
+
+	for i := range testReaderCases {
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			writeFd, err := ioutil.TempFile("", "TestReader_Live")
+			testutil.Ok(t, err)
+			defer os.Remove(writeFd.Name())
+
+			go func(i int) {
+				for _, rec := range testReaderCases[i].t {
+					rec := encodedRecord(rec.t, rec.b)
+					_, err := writeFd.Write(rec)
+					testutil.Ok(t, err)
+					runtime.Gosched()
+				}
+				writeFd.Close()
+			}(i)
+
+			// Read from a second FD on the same file.
+			readFd, err := os.Open(writeFd.Name())
+			testutil.Ok(t, err)
+			reader := NewLiveReader(logger, readFd)
+			for _, exp := range testReaderCases[i].exp {
+				for !reader.Next() {
+					testutil.Assert(t, reader.Err() == io.EOF, "expect EOF, got: %v", reader.Err())
+					runtime.Gosched()
+				}
+
+				actual := reader.Record()
+				testutil.Equals(t, exp, actual, "read wrong record")
+			}
+
+			testutil.Assert(t, !reader.Next(), "unexpected record")
+			if testReaderCases[i].fail {
+				testutil.Assert(t, reader.Err() != nil, "expected error")
+			}
+		})
+	}
+}
+
+const fuzzLen = 500
+
+func generateRandomEntries(w *WAL, records chan []byte) error {
+	var recs [][]byte
+	for i := 0; i < fuzzLen; i++ {
+		var sz int64
+		switch i % 5 {
+		case 0, 1:
+			sz = 50
+		case 2, 3:
+			sz = pageSize
+		default:
+			sz = pageSize * 8
+		}
+
+		rec := make([]byte, rand.Int63n(sz))
+		if _, err := rand.Read(rec); err != nil {
+			return err
+		}
+
+		records <- rec
+
+		// Randomly batch up records.
+		recs = append(recs, rec)
+		if rand.Intn(4) < 3 {
+			if err := w.Log(recs...); err != nil {
+				return err
+			}
+			recs = recs[:0]
+		}
+	}
+	return w.Log(recs...)
+}
+
+func allSegments(dir string) (io.Reader, error) {
+	seg, err := listSegments(dir)
+	if err != nil {
+		return nil, err
+	}
+
+	var readers []io.Reader
+	for _, r := range seg {
+		f, err := os.Open(filepath.Join(dir, r.name))
+		if err != nil {
+			return nil, err
+		}
+		readers = append(readers, f)
+	}
+	return io.MultiReader(readers...), nil
+}
+
+func TestReaderFuzz(t *testing.T) {
+	for name, fn := range readerConstructors {
+		t.Run(name, func(t *testing.T) {
+			dir, err := ioutil.TempDir("", "wal_fuzz_live")
+			testutil.Ok(t, err)
+			defer os.RemoveAll(dir)
+
+			w, err := NewSize(nil, nil, dir, 128*pageSize)
+			testutil.Ok(t, err)
+
+			// Buffering required as we're not reading concurrently.
+			input := make(chan []byte, fuzzLen)
+			err = generateRandomEntries(w, input)
+			testutil.Ok(t, err)
+			close(input)
+
+			err = w.Close()
+			testutil.Ok(t, err)
+
+			sr, err := allSegments(w.Dir())
+			testutil.Ok(t, err)
+
+			reader := fn(sr)
+			for expected := range input {
+				testutil.Assert(t, reader.Next(), "expected record: %v", reader.Err())
+				testutil.Equals(t, expected, reader.Record(), "read wrong record")
+			}
+			testutil.Assert(t, !reader.Next(), "unexpected record")
+		})
+	}
+}
+
+func TestReaderFuzz_Live(t *testing.T) {
+	logger := testutil.NewLogger(t)
+	dir, err := ioutil.TempDir("", "wal_fuzz_live")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	w, err := NewSize(nil, nil, dir, 128*pageSize)
+	testutil.Ok(t, err)
+
+	// In the background, generate a stream of random records and write them
+	// to the WAL.
+	input := make(chan []byte, fuzzLen/10) // buffering required as we sometimes batch WAL writes.
+	done := make(chan struct{})
+	go func() {
+		err := generateRandomEntries(w, input)
+		testutil.Ok(t, err)
+		time.Sleep(100 * time.Millisecond)
+		close(done)
+	}()
+
+	// Tail the WAL and compare the results.
+	m, _, err := w.Segments()
+	testutil.Ok(t, err)
+
+	seg, err := OpenReadSegment(SegmentName(dir, m))
+	testutil.Ok(t, err)
+
+	r := NewLiveReader(logger, seg)
+	segmentTicker := time.NewTicker(100 * time.Millisecond)
+	readTicker := time.NewTicker(10 * time.Millisecond)
+
+	readSegment := func(r *LiveReader) bool {
+		for r.Next() {
+			rec := r.Record()
+			expected, ok := <-input
+			testutil.Assert(t, ok, "unexpected record")
+			testutil.Equals(t, expected, rec, "record does not match expected")
+		}
+		testutil.Assert(t, r.Err() == io.EOF, "expected EOF, got: %v", r.Err())
+		return true
+	}
+
+outer:
+	for {
+		select {
+		case <-segmentTicker.C:
+			// check if new segments exist
+			_, last, err := w.Segments()
+			testutil.Ok(t, err)
+			if last <= seg.i {
+				continue
+			}
+
+			// read to end of segment.
+			readSegment(r)
+
+			fi, err := os.Stat(SegmentName(dir, seg.i))
+			testutil.Ok(t, err)
+			testutil.Assert(t, r.Offset() == fi.Size(), "expected to have read whole segment, but read %d of %d", r.Offset(), fi.Size())
+
+			seg, err = OpenReadSegment(SegmentName(dir, seg.i+1))
+			testutil.Ok(t, err)
+			r = NewLiveReader(logger, seg)
+
+		case <-readTicker.C:
+			readSegment(r)
+
+		case <-done:
+			readSegment(r)
+			break outer
+		}
+	}
+
+	testutil.Assert(t, r.Err() == io.EOF, "expected EOF")
+}
+
+func TestLiveReaderCorrupt_ShortFile(t *testing.T) {
+	// Write a corrupt WAL segment, there is one record of pageSize in length,
+	// but the segment is only half written.
+	logger := testutil.NewLogger(t)
+	dir, err := ioutil.TempDir("", "wal_live_corrupt")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	w, err := NewSize(nil, nil, dir, pageSize)
+	testutil.Ok(t, err)
+
+	rec := make([]byte, pageSize-recordHeaderSize)
+	_, err = rand.Read(rec)
+	testutil.Ok(t, err)
+
+	err = w.Log(rec)
+	testutil.Ok(t, err)
+
+	err = w.Close()
+	testutil.Ok(t, err)
+
+	segmentFile, err := os.OpenFile(filepath.Join(dir, "00000000"), os.O_RDWR, 0666)
+	testutil.Ok(t, err)
+
+	err = segmentFile.Truncate(pageSize / 2)
+	testutil.Ok(t, err)
+
+	err = segmentFile.Close()
+	testutil.Ok(t, err)
+
+	// Try and LiveReader it.
+	m, _, err := w.Segments()
+	testutil.Ok(t, err)
+
+	seg, err := OpenReadSegment(SegmentName(dir, m))
+	testutil.Ok(t, err)
+
+	r := NewLiveReader(logger, seg)
+	testutil.Assert(t, r.Next() == false, "expected no records")
+	testutil.Assert(t, r.Err() == io.EOF, "expected error, got: %v", r.Err())
+}
+
+func TestLiveReaderCorrupt_RecordTooLongAndShort(t *testing.T) {
+	// Write a corrupt WAL segment, when record len > page size.
+	logger := testutil.NewLogger(t)
+	dir, err := ioutil.TempDir("", "wal_live_corrupt")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	w, err := NewSize(nil, nil, dir, pageSize*2)
+	testutil.Ok(t, err)
+
+	rec := make([]byte, pageSize-recordHeaderSize)
+	_, err = rand.Read(rec)
+	testutil.Ok(t, err)
+
+	err = w.Log(rec)
+	testutil.Ok(t, err)
+
+	err = w.Close()
+	testutil.Ok(t, err)
+
+	segmentFile, err := os.OpenFile(filepath.Join(dir, "00000000"), os.O_RDWR, 0666)
+	testutil.Ok(t, err)
+
+	// Override the record length
+	buf := make([]byte, 3)
+	buf[0] = byte(recFull)
+	binary.BigEndian.PutUint16(buf[1:], 0xFFFF)
+	_, err = segmentFile.WriteAt(buf, 0)
+	testutil.Ok(t, err)
+
+	err = segmentFile.Close()
+	testutil.Ok(t, err)
+
+	// Try and LiveReader it.
+	m, _, err := w.Segments()
+	testutil.Ok(t, err)
+
+	seg, err := OpenReadSegment(SegmentName(dir, m))
+	testutil.Ok(t, err)
+
+	r := NewLiveReader(logger, seg)
+	testutil.Assert(t, r.Next() == false, "expected no records")
+	testutil.Assert(t, r.Err().Error() == "record length greater than a single page: 65542 > 32768", "expected error, got: %v", r.Err())
+}
+
+func TestReaderData(t *testing.T) {
+	dir := os.Getenv("WALDIR")
+	if dir == "" {
+		return
+	}
+
+	for name, fn := range readerConstructors {
+		t.Run(name, func(t *testing.T) {
+			w, err := New(nil, nil, dir)
+			testutil.Ok(t, err)
+
+			sr, err := allSegments(dir)
+			testutil.Ok(t, err)
+
+			reader := fn(sr)
+			for reader.Next() {
+			}
+			testutil.Ok(t, reader.Err())
+
+			err = w.Repair(reader.Err())
+			testutil.Ok(t, err)
+		})
+	}
+}
diff --git a/wal/wal_test.go b/wal/wal_test.go
index 737c44c..898030a 100644
--- a/wal/wal_test.go
+++ b/wal/wal_test.go
@@ -16,393 +16,16 @@ package wal
 
 import (
 	"bytes"
-	"encoding/binary"
 	"fmt"
-	"hash/crc32"
-	"io"
 	"io/ioutil"
 	"math/rand"
 	"os"
-	"path"
-	"sync"
+	"path/filepath"
 	"testing"
-	"time"
 
 	"github.com/prometheus/tsdb/testutil"
 )
 
-type record struct {
-	t recType
-	b []byte
-}
-
-var data = make([]byte, 100000)
-var testReaderCases = []struct {
-	t    []record
-	exp  [][]byte
-	fail bool
-}{
-	// Sequence of valid records.
-	{
-		t: []record{
-			{recFull, data[0:200]},
-			{recFirst, data[200:300]},
-			{recLast, data[300:400]},
-			{recFirst, data[400:800]},
-			{recMiddle, data[800:900]},
-			{recPageTerm, make([]byte, pageSize-900-recordHeaderSize*5-1)}, // exactly lines up with page boundary.
-			{recLast, data[900:900]},
-			{recFirst, data[900:1000]},
-			{recMiddle, data[1000:1200]},
-			{recMiddle, data[1200:30000]},
-			{recMiddle, data[30000:30001]},
-			{recMiddle, data[30001:30001]},
-			{recLast, data[30001:32000]},
-		},
-		exp: [][]byte{
-			data[0:200],
-			data[200:400],
-			data[400:900],
-			data[900:32000],
-		},
-	},
-	// Exactly at the limit of one page minus the header size
-	{
-		t: []record{
-			{recFull, data[0 : pageSize-recordHeaderSize]},
-		},
-		exp: [][]byte{
-			data[:pageSize-recordHeaderSize],
-		},
-	},
-	// More than a full page, this exceeds our buffer and can never happen
-	// when written by the WAL.
-	{
-		t: []record{
-			{recFull, data[0 : pageSize+1]},
-		},
-		fail: true,
-	},
-	// Invalid orders of record types.
-	{
-		t:    []record{{recMiddle, data[:200]}},
-		fail: true,
-	},
-	{
-		t:    []record{{recLast, data[:200]}},
-		fail: true,
-	},
-	{
-		t: []record{
-			{recFirst, data[:200]},
-			{recFull, data[200:400]},
-		},
-		fail: true,
-	},
-	{
-		t: []record{
-			{recFirst, data[:100]},
-			{recMiddle, data[100:200]},
-			{recFull, data[200:400]},
-		},
-		fail: true,
-	},
-	// Non-zero data after page termination.
-	{
-		t: []record{
-			{recFull, data[:100]},
-			{recPageTerm, append(make([]byte, 1000), 1)},
-		},
-		exp:  [][]byte{data[:100]},
-		fail: true,
-	},
-}
-
-func encodedRecord(t recType, b []byte) []byte {
-	if t == recPageTerm {
-		return append([]byte{0}, b...)
-	}
-	r := make([]byte, recordHeaderSize)
-	r[0] = byte(t)
-	binary.BigEndian.PutUint16(r[1:], uint16(len(b)))
-	binary.BigEndian.PutUint32(r[3:], crc32.Checksum(b, castagnoliTable))
-	return append(r, b...)
-}
-
-// TestReader feeds the reader a stream of encoded records with different types.
-func TestReader(t *testing.T) {
-	for i, c := range testReaderCases {
-		t.Logf("test %d", i)
-
-		var buf []byte
-		for _, r := range c.t {
-			buf = append(buf, encodedRecord(r.t, r.b)...)
-		}
-		r := NewReader(bytes.NewReader(buf))
-
-		for j := 0; r.Next(); j++ {
-			t.Logf("record %d", j)
-			rec := r.Record()
-
-			if j >= len(c.exp) {
-				t.Fatal("received more records than inserted")
-			}
-			testutil.Equals(t, c.exp[j], rec, "Bytes within record did not match expected Bytes")
-		}
-		if !c.fail && r.Err() != nil {
-			t.Fatalf("unexpected error: %s", r.Err())
-		}
-		if c.fail && r.Err() == nil {
-			t.Fatalf("expected error but got none")
-		}
-	}
-}
-
-func TestReader_Live(t *testing.T) {
-	for i, c := range testReaderCases {
-		t.Logf("test %d", i)
-		dir, err := ioutil.TempDir("", fmt.Sprintf("live_reader_%d", i))
-		t.Logf("created dir %s", dir)
-		testutil.Ok(t, err)
-		defer os.RemoveAll(dir)
-
-		// we're never going to have more than a single segment file per test case right now
-		f, err := os.Create(path.Join(dir, "00000000"))
-		testutil.Ok(t, err)
-
-		// live reader doesn't work on readers created from bytes buffers,
-		// since we need to be able to write more data to the thing we're
-		// reading from after the reader has been created
-		wg := sync.WaitGroup{}
-		// make sure the reader doesn't start until at least one record is written
-		wg.Add(1)
-		go func() {
-			for i, rec := range c.t {
-				rec := encodedRecord(rec.t, rec.b)
-				n, err := f.Write(rec)
-				testutil.Ok(t, err)
-				testutil.Assert(t, n > 0, "no bytes were written to wal")
-				if i == 0 {
-					wg.Done()
-				}
-			}
-		}()
-		sr, err := OpenReadSegment(SegmentName(dir, 0))
-		testutil.Ok(t, err)
-		lr := NewLiveReader(sr)
-		j := 0
-		wg.Wait()
-	caseLoop:
-		for {
-			for ; lr.Next(); j++ {
-				rec := lr.Record()
-				t.Log("j: ", j)
-				testutil.Equals(t, c.exp[j], rec, "Bytes within record did not match expected Bytes")
-				if j == len(c.exp)-1 {
-					break caseLoop
-				}
-
-			}
-
-			// Because reads and writes are happening concurrently, unless we get an error we should
-			// attempt to read records again.
-			if j == 0 && lr.Err() == nil {
-				continue
-			}
-
-			if !c.fail && lr.Err() != nil {
-				t.Fatalf("unexpected error: %s", lr.Err())
-			}
-			if c.fail && lr.Err() == nil {
-				t.Fatalf("expected error but got none:\n\tinput: %+v", c.t)
-			}
-			if lr.Err() != nil {
-				t.Log("err: ", lr.Err())
-				break
-			}
-		}
-	}
-}
-
-func TestWAL_FuzzWriteRead_Live(t *testing.T) {
-	const count = 5000
-	var input [][]byte
-	lock := sync.RWMutex{}
-	var recs [][]byte
-	var index int
-
-	// Get size of segment.
-	getSegmentSize := func(dir string, index int) (int64, error) {
-		i := int64(-1)
-		fi, err := os.Stat(SegmentName(dir, index))
-		if err == nil {
-			i = fi.Size()
-		}
-		return i, err
-	}
-
-	readSegment := func(r *LiveReader) {
-		for r.Next() {
-			rec := r.Record()
-			lock.RLock()
-			l := len(input)
-			lock.RUnlock()
-			if index >= l {
-				t.Fatalf("read too many records")
-			}
-			lock.RLock()
-			if !bytes.Equal(input[index], rec) {
-				t.Fatalf("record %d (len %d) does not match (expected len %d)",
-					index, len(rec), len(input[index]))
-			}
-			lock.RUnlock()
-			index++
-		}
-		if r.Err() != io.EOF {
-			testutil.Ok(t, r.Err())
-		}
-	}
-
-	dir, err := ioutil.TempDir("", "wal_fuzz_live")
-	t.Log("created dir: ", dir)
-	testutil.Ok(t, err)
-	defer func() {
-		os.RemoveAll(dir)
-	}()
-
-	w, err := NewSize(nil, nil, dir, 128*pageSize)
-	testutil.Ok(t, err)
-
-	go func() {
-		for i := 0; i < count; i++ {
-			var sz int64
-			switch i % 5 {
-			case 0, 1:
-				sz = 50
-			case 2, 3:
-				sz = pageSize
-			default:
-				sz = pageSize * 8
-			}
-
-			rec := make([]byte, rand.Int63n(sz))
-			_, err := rand.Read(rec)
-			testutil.Ok(t, err)
-			lock.Lock()
-			input = append(input, rec)
-			lock.Unlock()
-			recs = append(recs, rec)
-
-			// Randomly batch up records.
-			if rand.Intn(4) < 3 {
-				testutil.Ok(t, w.Log(recs...))
-				recs = recs[:0]
-			}
-		}
-		testutil.Ok(t, w.Log(recs...))
-	}()
-
-	m, _, err := w.Segments()
-	testutil.Ok(t, err)
-
-	seg, err := OpenReadSegment(SegmentName(dir, m))
-	testutil.Ok(t, err)
-
-	r := NewLiveReader(seg)
-	segmentTicker := time.NewTicker(100 * time.Millisecond)
-	readTicker := time.NewTicker(10 * time.Millisecond)
-	for {
-		select {
-		case <-segmentTicker.C:
-			// check if new segments exist
-			_, last, err := w.Segments()
-			testutil.Ok(t, err)
-			if last > seg.i {
-				for {
-					readSegment(r)
-					if r.Err() != io.EOF {
-						testutil.Ok(t, r.Err())
-					}
-					size, err := getSegmentSize(dir, seg.i)
-					testutil.Ok(t, err)
-					// make sure we've read all of the current segment before rotating
-					if r.TotalRead() == size {
-						break
-					}
-				}
-				seg, err = OpenReadSegment(SegmentName(dir, seg.i+1))
-				testutil.Ok(t, err)
-				r = NewLiveReader(seg)
-			}
-		case <-readTicker.C:
-			readSegment(r)
-		}
-		if index == count {
-			break
-		}
-	}
-	testutil.Ok(t, r.Err())
-}
-func TestWAL_FuzzWriteRead(t *testing.T) {
-	const count = 25000
-
-	dir, err := ioutil.TempDir("", "walfuzz")
-	testutil.Ok(t, err)
-	defer os.RemoveAll(dir)
-
-	w, err := NewSize(nil, nil, dir, 128*pageSize)
-	testutil.Ok(t, err)
-
-	var input [][]byte
-	var recs [][]byte
-
-	for i := 0; i < count; i++ {
-		var sz int
-		switch i % 5 {
-		case 0, 1:
-			sz = 50
-		case 2, 3:
-			sz = pageSize
-		default:
-			sz = 8 * pageSize
-		}
-		rec := make([]byte, rand.Intn(sz))
-		_, err := rand.Read(rec)
-		testutil.Ok(t, err)
-
-		input = append(input, rec)
-		recs = append(recs, rec)
-
-		// Randomly batch up records.
-		if rand.Intn(4) < 3 {
-			testutil.Ok(t, w.Log(recs...))
-			recs = recs[:0]
-		}
-	}
-	testutil.Ok(t, w.Log(recs...))
-
-	m, n, err := w.Segments()
-	testutil.Ok(t, err)
-
-	rc, err := NewSegmentsRangeReader(SegmentRange{Dir: dir, First: m, Last: n})
-	testutil.Ok(t, err)
-	defer rc.Close()
-
-	rdr := NewReader(rc)
-
-	for i := 0; rdr.Next(); i++ {
-		rec := rdr.Record()
-		if i >= len(input) {
-			t.Fatal("read too many records")
-		}
-		if !bytes.Equal(input[i], rec) {
-			t.Fatalf("record %d (len %d) does not match (expected len %d)",
-				i, len(rec), len(input[i]))
-		}
-	}
-	testutil.Ok(t, rdr.Err())
-}
-
 func TestWAL_Repair(t *testing.T) {
 
 	for name, test := range map[string]struct {
@@ -547,6 +170,129 @@ func TestWAL_Repair(t *testing.T) {
 	}
 }
 
+// TestCorruptAndCarryOn writes a multi-segment WAL; corrupts the first segment and
+// ensures that an error during reading that segment are correctly repaired before
+// moving to write more records to the WAL.
+func TestCorruptAndCarryOn(t *testing.T) {
+	dir, err := ioutil.TempDir("", "wal_repair")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	var (
+		logger      = testutil.NewLogger(t)
+		segmentSize = pageSize * 3
+		recordSize  = (pageSize / 3) - recordHeaderSize
+	)
+
+	// Produce a WAL with a two segments of 3 pages with 3 records each,
+	// so when we truncate the file we're guaranteed to split a record.
+	{
+		w, err := NewSize(logger, nil, dir, segmentSize)
+		testutil.Ok(t, err)
+
+		for i := 0; i < 18; i++ {
+			buf := make([]byte, recordSize)
+			_, err := rand.Read(buf)
+			testutil.Ok(t, err)
+
+			err = w.Log(buf)
+			testutil.Ok(t, err)
+		}
+
+		err = w.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Check all the segments are the correct size.
+	{
+		segments, err := listSegments(dir)
+		testutil.Ok(t, err)
+		for _, segment := range segments {
+			f, err := os.OpenFile(filepath.Join(dir, fmt.Sprintf("%08d", segment.index)), os.O_RDONLY, 0666)
+			testutil.Ok(t, err)
+
+			fi, err := f.Stat()
+			testutil.Ok(t, err)
+
+			t.Log("segment", segment.index, "size", fi.Size())
+			testutil.Equals(t, int64(segmentSize), fi.Size())
+
+			err = f.Close()
+			testutil.Ok(t, err)
+		}
+	}
+
+	// Truncate the first file, splitting the middle record in the second
+	// page in half, leaving 4 valid records.
+	{
+		f, err := os.OpenFile(filepath.Join(dir, fmt.Sprintf("%08d", 0)), os.O_RDWR, 0666)
+		testutil.Ok(t, err)
+
+		fi, err := f.Stat()
+		testutil.Ok(t, err)
+		testutil.Equals(t, int64(segmentSize), fi.Size())
+
+		err = f.Truncate(int64(segmentSize / 2))
+		testutil.Ok(t, err)
+
+		err = f.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Now try and repair this WAL, and write 5 more records to it.
+	{
+		sr, err := NewSegmentsReader(dir)
+		testutil.Ok(t, err)
+
+		reader := NewReader(sr)
+		i := 0
+		for ; i < 4 && reader.Next(); i++ {
+			testutil.Equals(t, recordSize, len(reader.Record()))
+		}
+		testutil.Equals(t, 4, i, "not enough records")
+		testutil.Assert(t, !reader.Next(), "unexpected record")
+
+		corruptionErr := reader.Err()
+		testutil.Assert(t, corruptionErr != nil, "expected error")
+
+		err = sr.Close()
+		testutil.Ok(t, err)
+
+		w, err := NewSize(logger, nil, dir, segmentSize)
+		testutil.Ok(t, err)
+
+		err = w.Repair(corruptionErr)
+		testutil.Ok(t, err)
+
+		for i := 0; i < 5; i++ {
+			buf := make([]byte, recordSize)
+			_, err := rand.Read(buf)
+			testutil.Ok(t, err)
+
+			err = w.Log(buf)
+			testutil.Ok(t, err)
+		}
+
+		err = w.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Replay the WAL. Should get 9 records.
+	{
+		sr, err := NewSegmentsReader(dir)
+		testutil.Ok(t, err)
+
+		reader := NewReader(sr)
+		i := 0
+		for ; i < 9 && reader.Next(); i++ {
+			testutil.Equals(t, recordSize, len(reader.Record()))
+		}
+		testutil.Equals(t, 9, i, "wrong number of records")
+		testutil.Assert(t, !reader.Next(), "unexpected record")
+		testutil.Equals(t, nil, reader.Err())
+	}
+}
+
 func BenchmarkWAL_LogBatched(b *testing.B) {
 	dir, err := ioutil.TempDir("", "bench_logbatch")
 	testutil.Ok(b, err)