jldb/lib/wal/segment_test.go

package wal

import (
	"bytes"
	crand "crypto/rand"
	"io"
	"path/filepath"
	"testing"
	"time"

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

func newSegmentForTesting(t *testing.T) *segment {
	tmpDir := t.TempDir()
	seg, err := createSegment(filepath.Join(tmpDir, "x"), 1, 100, 200)
	if err != nil {
		t.Fatal(err)
	}
	return seg
}

func TestNewSegmentDirNotFound(t *testing.T) {
	t.Parallel()
	tmpDir := t.TempDir()
	p := filepath.Join(tmpDir, "notFound", "1245")

	if _, err := createSegment(p, 1, 1234, 5678); !errs.IO.Is(err) {
		t.Fatal(err)
	}
}

func TestOpenSegmentNotFound(t *testing.T) {
	t.Parallel()
	tmpDir := t.TempDir()
	p := filepath.Join(tmpDir, "notFound")

	if _, err := openSegment(p, 1); !errs.IO.Is(err) {
		t.Fatal(err)
	}
}

func TestOpenSegmentTruncatedFile(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	path := seg.f.Name()
	if err := seg.f.Truncate(4); err != nil {
		t.Fatal(err)
	}
	seg.Close()

	if _, err := openSegment(path, 1); !errs.IO.Is(err) {
		t.Fatal(err)
	}
}

func TestOpenSegmentCorruptHeader(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	path := seg.f.Name()
	buf := make([]byte, atomicheader.ReservedBytes)
	crand.Read(buf)

	if _, err := seg.f.Seek(0, io.SeekStart); err != nil {
		t.Fatal(err)
	}

	if _, err := seg.f.Write(buf); err != nil {
		t.Fatal(err)
	}

	seg.Close()

	if _, err := openSegment(path, 1); !errs.Corrupt.Is(err) {
		t.Fatal(err)
	}
}

func TestOpenSegmentCorruptHeader2(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	path := seg.f.Name()
	buf := make([]byte, 1024) // 2 pages.
	crand.Read(buf)

	if _, err := seg.f.Seek(1024, io.SeekStart); err != nil {
		t.Fatal(err)
	}

	if _, err := seg.f.Write(buf); err != nil {
		t.Fatal(err)
	}

	seg.Close()

	if _, err := openSegment(path, 1); !errs.Corrupt.Is(err) {
		t.Fatal(err)
	}
}

func TestSegmentArchiveTwice(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	for i := 0; i < 2; i++ {
		if err := seg.Archive(); err != nil {
			t.Fatal(err)
		}
	}
}

func TestSegmentAppendArchived(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	appendRandomRecords(t, seg, 8)

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

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

func TestSegmentAppendRecordInvalidSeqNum(t *testing.T) {
	t.Parallel()
	seg := newSegmentForTesting(t)

	appendRandomRecords(t, seg, 8) // 109 is next.

	_, _, err := seg.appendRecord(Record{
		SeqNum:      110,
		TimestampMS: time.Now().UnixMilli(),
		DataSize:    100,
	})
	if !errs.Unexpected.Is(err) {
		t.Fatal(err)
	}
}