Backport corrupted WAL error fix

tail/tail.go

@@ -207,19 +207,19 @@ func (t *Tailer) Read(b []byte) (int, error) {
 	}
 }
 
+// openSegment finds a WAL segment with a name that parses to n. This
+// way we do not need to know how wide the segment filename is (i.e.,
+// how many zeros to pad with).
 func openSegment(dir string, n int) (io.ReadCloser, error) {
 	files, err := fileutil.ReadDir(dir)
 	if err != nil {
 		return nil, err
 	}
 	for _, fn := range files {
 		k, err := strconv.Atoi(fn)
-		if err != nil || k < n {
+		if err != nil || k != n {
 			continue
 		}
-		if k > n {
-			return nil, errors.Errorf("next segment %d too high, expected %d", n, k)
-		}
 		return wal.OpenReadSegment(filepath.Join(dir, fn))
 	}
 	return nil, tsdb.ErrNotFound

tail/tail_test.go

@@ -17,15 +17,57 @@ package tail
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io/ioutil"
 	"math/rand"
 	"os"
+	"path"
 	"testing"
 	"time"
 
 	"github.com/prometheus/tsdb/wal"
 )
 
+func TestOpenSegment(t *testing.T) {
+	dir, err := ioutil.TempDir("", "test_open_segment")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	if err := ioutil.WriteFile(path.Join(dir, "000000000000000000000nonsense"), []byte("bad"), 0777); err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 10; i++ {
+		if err := ioutil.WriteFile(path.Join(dir, fmt.Sprint("000000000000000000000", i)), []byte(fmt.Sprint(i)), 0777); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for i := 19; i >= 10; i-- {
+		if err := ioutil.WriteFile(path.Join(dir, fmt.Sprint("000000000000000000000", i)), []byte(fmt.Sprint(i)), 0777); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < 20; i++ {
+		rc, err := openSegment(dir, i)
+		if err != nil {
+			t.Fatal(err)
+		}
+		body, err := ioutil.ReadAll(rc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if want, have := fmt.Sprint(i), string(body); want != have {
+			t.Fatalf("invalid body read want=%q have=%q", want, have)
+		}
+		if err := rc.Close(); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
 func TestTailFuzz(t *testing.T) {
 	dir, err := ioutil.TempDir("", "test_tail")
 	if err != nil {