jldb/lib/wal/generic_test.go

package wal

import (
	"bytes"
	"encoding/binary"
	"errors"
	"io"
	"math/rand"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	"git.crumpington.com/public/jldb/lib/errs"
)

type waLog interface {
	Append(int64, io.Reader) (int64, int64, error)
	appendRecord(Record) (int64, int64, error)
	Iterator(int64) (Iterator, error)
	Close() error
}

func TestGenericTestHarness_segment(t *testing.T) {
	t.Parallel()
	(&GenericTestHarness{
		New: func(tmpDir string, firstSeqNum int64) (waLog, error) {
			l, err := createSegment(filepath.Join(tmpDir, "x"), 1, firstSeqNum, 12345)
			return l, err
		},
	}).Run(t)
}

func TestGenericTestHarness_wal(t *testing.T) {
	t.Parallel()
	(&GenericTestHarness{
		New: func(tmpDir string, firstSeqNum int64) (waLog, error) {
			l, err := Create(tmpDir, firstSeqNum, Config{
				SegMinCount:  1,
				SegMaxAgeSec: 1,
			})
			return l, err
		},
	}).Run(t)
}

// ----------------------------------------------------------------------------

type GenericTestHarness struct {
	New func(tmpDir string, firstSeqNum int64) (waLog, error)
}

func (h *GenericTestHarness) Run(t *testing.T) {
	val := reflect.ValueOf(h)
	typ := val.Type()
	for i := 0; i < typ.NumMethod(); i++ {
		method := typ.Method(i)

		if !strings.HasPrefix(method.Name, "Test") {
			continue
		}

		t.Run(method.Name, func(t *testing.T) {
			t.Parallel()
			firstSeqNum := rand.Int63n(23423)

			tmpDir := t.TempDir()

			wal, err := h.New(tmpDir, firstSeqNum)
			if err != nil {
				t.Fatal(err)
			}
			defer wal.Close()

			val.MethodByName(method.Name).Call([]reflect.Value{
				reflect.ValueOf(t),
				reflect.ValueOf(firstSeqNum),
				reflect.ValueOf(wal),
			})
		})
	}
}

// ----------------------------------------------------------------------------

func (h *GenericTestHarness) TestBasic(t *testing.T, firstSeqNum int64, wal waLog) {
	expected := appendRandomRecords(t, wal, 123)

	for i := 0; i < 123; i++ {
		it, err := wal.Iterator(firstSeqNum + int64(i))
		if err != nil {
			t.Fatal(err)
		}

		checkIteratorMatches(t, it, expected[i:])

		it.Close()
	}
}

func (h *GenericTestHarness) TestAppendNotFound(t *testing.T, firstSeqNum int64, wal waLog) {
	recs := appendRandomRecords(t, wal, 123)
	lastSeqNum := recs[len(recs)-1].SeqNum

	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	it.Close()

	it, err = wal.Iterator(lastSeqNum + 1)
	if err != nil {
		t.Fatal(err)
	}
	it.Close()

	if _, err = wal.Iterator(firstSeqNum - 1); !errs.NotFound.Is(err) {
		t.Fatal(err)
	}

	if _, err = wal.Iterator(lastSeqNum + 2); !errs.NotFound.Is(err) {
		t.Fatal(err)
	}
}

func (h *GenericTestHarness) TestNextAfterClose(t *testing.T, firstSeqNum int64, wal waLog) {
	appendRandomRecords(t, wal, 123)

	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it.Close()

	if !it.Next(0) {
		t.Fatal("Should be next")
	}

	if err := wal.Close(); err != nil {
		t.Fatal(err)
	}

	if it.Next(0) {
		t.Fatal("Shouldn't be next")
	}

	if !errs.Closed.Is(it.Error()) {
		t.Fatal(it.Error())
	}
}

func (h *GenericTestHarness) TestNextTimeout(t *testing.T, firstSeqNum int64, wal waLog) {
	recs := appendRandomRecords(t, wal, 123)

	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it.Close()

	for range recs {
		if !it.Next(0) {
			t.Fatal("Expected next")
		}
	}

	if it.Next(time.Millisecond) {
		t.Fatal("Unexpected next")
	}
}

func (h *GenericTestHarness) TestNextNotify(t *testing.T, firstSeqNum int64, wal waLog) {
	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it.Close()

	recsC := make(chan []RawRecord, 1)

	go func() {
		time.Sleep(time.Second)
		recsC <- appendRandomRecords(t, wal, 1)
	}()

	if !it.Next(time.Hour) {
		t.Fatal("expected next")
	}

	recs := <-recsC
	rec := it.Record()
	if rec.SeqNum != recs[0].SeqNum {
		t.Fatal(rec)
	}
}

func (h *GenericTestHarness) TestNextArchived(t *testing.T, firstSeqNum int64, wal waLog) {
	type archiver interface {
		Archive() error
	}

	arch, ok := wal.(archiver)
	if !ok {
		return
	}

	recs := appendRandomRecords(t, wal, 10)

	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it.Close()

	if err := arch.Archive(); err != nil {
		t.Fatal(err)
	}

	for i, expected := range recs {
		if !it.Next(time.Millisecond) {
			t.Fatal(i, "no next")
		}

		rec := it.Record()
		if rec.SeqNum != expected.SeqNum {
			t.Fatal(rec, expected)
		}
	}

	if it.Next(time.Minute) {
		t.Fatal("unexpected next")
	}

	if !errs.EOFArchived.Is(it.Error()) {
		t.Fatal(it.Error())
	}
}

func (h *GenericTestHarness) TestWriteReadConcurrent(t *testing.T, firstSeqNum int64, wal waLog) {
	N := 1200

	writeErr := make(chan error, 1)

	dataSize := int64(4)
	makeData := func(i int) []byte {
		data := make([]byte, 4)
		binary.LittleEndian.PutUint32(data, uint32(i))
		return data
	}

	go func() {
		for i := 0; i < N; i++ {

			seqNum, _, err := wal.Append(dataSize, bytes.NewBuffer(makeData(i)))
			if err != nil {
				writeErr <- err
				return
			}

			if seqNum != int64(i)+firstSeqNum {
				writeErr <- errors.New("Incorrect seq num")
				return
			}

			time.Sleep(time.Millisecond)
		}

		writeErr <- nil
	}()

	it, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it.Close()

	for i := 0; i < N; i++ {
		if !it.Next(time.Minute) {
			t.Fatal("expected next", i, it.Error(), it.Record())
		}

		expectedData := makeData(i)
		rec := it.Record()

		data, err := io.ReadAll(rec.Reader)
		if err != nil {
			t.Fatal(err)
		}

		if !bytes.Equal(data, expectedData) {
			t.Fatal(data, expectedData)
		}
	}

	if err := <-writeErr; err != nil {
		t.Fatal(err)
	}
}

func (h *GenericTestHarness) TestAppendAfterClose(t *testing.T, firstSeqNum int64, wal waLog) {
	if _, _, err := wal.Append(4, bytes.NewBuffer([]byte{1, 2, 3, 4})); err != nil {
		t.Fatal(err)
	}

	wal.Close()

	_, _, err := wal.Append(4, bytes.NewBuffer([]byte{1, 2, 3, 4}))
	if !errs.Closed.Is(err) {
		t.Fatal(err)
	}
}

func (h *GenericTestHarness) TestIterateNegativeOne(t *testing.T, firstSeqNum int64, wal waLog) {
	recs := appendRandomRecords(t, wal, 10)

	it1, err := wal.Iterator(firstSeqNum)
	if err != nil {
		t.Fatal(err)
	}
	defer it1.Close()

	it2, err := wal.Iterator(-1)
	if err != nil {
		t.Fatal(err)
	}
	defer it2.Close()

	if !it1.Next(0) {
		t.Fatal(0)
	}
	if !it2.Next(0) {
		t.Fatal(0)
	}

	r1 := it1.Record()
	r2 := it2.Record()

	if r1.SeqNum != r2.SeqNum || r1.SeqNum != firstSeqNum || r1.SeqNum != recs[0].SeqNum {
		t.Fatal(r1.SeqNum, r2.SeqNum, firstSeqNum, recs[0].SeqNum)
	}
}

func (h *GenericTestHarness) TestIteratorAfterClose(t *testing.T, firstSeqNum int64, wal waLog) {
	appendRandomRecords(t, wal, 10)
	wal.Close()

	if _, err := wal.Iterator(-1); !errs.Closed.Is(err) {
		t.Fatal(err)
	}
}

func (h *GenericTestHarness) TestIteratorNextWithError(t *testing.T, firstSeqNum int64, wal waLog) {
	appendRandomRecords(t, wal, 10)
	it, err := wal.Iterator(-1)
	if err != nil {
		t.Fatal(err)
	}

	wal.Close()

	it.Next(0)
	if !errs.Closed.Is(it.Error()) {
		t.Fatal(it.Error())
	}

	it.Next(0)
	if !errs.Closed.Is(it.Error()) {
		t.Fatal(it.Error())
	}
}

func (h *GenericTestHarness) TestIteratorConcurrentClose(t *testing.T, firstSeqNum int64, wal waLog) {
	it, err := wal.Iterator(-1)
	if err != nil {
		t.Fatal(err)
	}

	go func() {
		writeRandomWithEOF(wal, 3*time.Second)
		wal.Close()
	}()

	for it.Next(time.Hour) {
		// Skip.
	}

	// Error may be Closed or NotFound.
	if !errs.Closed.Is(it.Error()) && !errs.NotFound.Is(it.Error()) {
		t.Fatal(it.Error())
	}
}