seaweedfs/weed/shell/command_fs_meta_load_test.go

package shell

import (
	"bytes"
	"context"
	"errors"
	"io"
	"sync"
	"testing"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/protobuf/proto"

	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/util"
)

type stubMetaLoadClient struct {
	filer_pb.SeaweedFilerClient
	createEntry func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error)
}

func (s *stubMetaLoadClient) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest, opts ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) {
	return s.createEntry(ctx, req)
}

func writeMetaEntry(t *testing.T, buf *bytes.Buffer, dir, name string, isDir bool) {
	t.Helper()
	data, err := proto.Marshal(&filer_pb.FullEntry{
		Dir:   dir,
		Entry: &filer_pb.Entry{Name: name, IsDirectory: isDir},
	})
	if err != nil {
		t.Fatalf("marshal entry %s/%s: %v", dir, name, err)
	}
	sizeBuf := make([]byte, 4)
	util.Uint32toBytes(sizeBuf, uint32(len(data)))
	buf.Write(sizeBuf)
	buf.Write(data)
}

func runLoadEntries(c *commandFsMetaLoad, buf *bytes.Buffer, client filer_pb.SeaweedFilerClient) (dirCount, fileCount uint64, err error) {
	return c.loadEntries(buf, client, "", 4, false, io.Discard)
}

func TestFsMetaLoadCreatesAllEntries(t *testing.T) {
	var buf bytes.Buffer
	writeMetaEntry(t, &buf, "/dir", "sub", true)
	writeMetaEntry(t, &buf, "/dir", "file1", false)
	writeMetaEntry(t, &buf, "/dir", "file2", false)

	var mu sync.Mutex
	var created []string
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		mu.Lock()
		created = append(created, req.Directory+"/"+req.Entry.Name)
		mu.Unlock()
		return &filer_pb.CreateEntryResponse{}, nil
	}}

	c := &commandFsMetaLoad{}
	dirCount, fileCount, err := runLoadEntries(c, &buf, client)
	if err != nil {
		t.Fatalf("loadEntries: %v", err)
	}
	if dirCount != 1 || fileCount != 2 {
		t.Fatalf("got %d dirs %d files, want 1 dir 2 files", dirCount, fileCount)
	}
	if len(created) != 3 {
		t.Fatalf("got %d CreateEntry calls, want 3", len(created))
	}
}

func TestFsMetaLoadWaitsForInflightCreateAtEOF(t *testing.T) {
	var buf bytes.Buffer
	writeMetaEntry(t, &buf, "/dir", "file1", false)

	started := make(chan struct{})
	release := make(chan struct{})
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		close(started)
		<-release
		return &filer_pb.CreateEntryResponse{}, nil
	}}

	c := &commandFsMetaLoad{}
	done := make(chan error, 1)
	go func() {
		_, _, err := runLoadEntries(c, &buf, client)
		done <- err
	}()

	select {
	case <-started:
	case <-time.After(5 * time.Second):
		t.Fatal("CreateEntry was not called")
	}

	select {
	case err := <-done:
		t.Fatalf("loadEntries returned %v while CreateEntry was still in flight", err)
	case <-time.After(100 * time.Millisecond):
	}

	close(release)
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("loadEntries: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("loadEntries did not return after CreateEntry completed")
	}
}

func TestFsMetaLoadWaitsForInflightCreateOnCorruptData(t *testing.T) {
	var buf bytes.Buffer
	writeMetaEntry(t, &buf, "/dir", "file1", false)
	writeCorruptTail(&buf)

	started := make(chan struct{})
	release := make(chan struct{})
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		close(started)
		<-release
		return &filer_pb.CreateEntryResponse{}, nil
	}}

	c := &commandFsMetaLoad{}
	done := make(chan error, 1)
	go func() {
		_, _, err := runLoadEntries(c, &buf, client)
		done <- err
	}()

	select {
	case <-started:
	case <-time.After(5 * time.Second):
		t.Fatal("CreateEntry was not called")
	}

	select {
	case err := <-done:
		t.Fatalf("loadEntries returned %v while CreateEntry was still in flight", err)
	case <-time.After(100 * time.Millisecond):
	}

	close(release)
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("loadEntries returned nil on truncated input")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("loadEntries did not return after CreateEntry completed")
	}
}

func TestFsMetaLoadReturnsWorkerError(t *testing.T) {
	var buf bytes.Buffer
	writeMetaEntry(t, &buf, "/dir", "file1", false)

	workerErr := errors.New("create entry rejected")
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		return nil, workerErr
	}}

	c := &commandFsMetaLoad{}
	_, _, err := runLoadEntries(c, &buf, client)
	if !errors.Is(err, workerErr) {
		t.Fatalf("got %v, want wrapped %v", err, workerErr)
	}
}

func TestFsMetaLoadPrefersWorkerErrorOverReadError(t *testing.T) {
	var buf bytes.Buffer
	writeMetaEntry(t, &buf, "/dir", "file1", false)
	writeCorruptTail(&buf)

	workerErr := errors.New("create entry rejected")
	started := make(chan struct{})
	release := make(chan struct{})
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		close(started)
		<-release
		return nil, workerErr
	}}

	c := &commandFsMetaLoad{}
	done := make(chan error, 1)
	go func() {
		_, _, err := runLoadEntries(c, &buf, client)
		done <- err
	}()

	select {
	case <-started:
	case <-time.After(5 * time.Second):
		t.Fatal("CreateEntry was not called")
	}
	close(release)

	select {
	case err := <-done:
		if !errors.Is(err, workerErr) {
			t.Fatalf("got %v, want wrapped %v", err, workerErr)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("loadEntries did not return")
	}
}

func TestFsMetaLoadReturnsErrorOnCorruptData(t *testing.T) {
	client := &stubMetaLoadClient{createEntry: func(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
		return &filer_pb.CreateEntryResponse{}, nil
	}}
	c := &commandFsMetaLoad{}

	t.Run("truncated entry data", func(t *testing.T) {
		var buf bytes.Buffer
		writeMetaEntry(t, &buf, "/dir", "file1", false)
		writeCorruptTail(&buf)
		if _, _, err := runLoadEntries(c, &buf, client); err == nil {
			t.Fatal("loadEntries returned nil on truncated input")
		}
	})

	t.Run("malformed proto", func(t *testing.T) {
		var buf bytes.Buffer
		writeMetaEntry(t, &buf, "/dir", "file1", false)
		sizeBuf := make([]byte, 4)
		util.Uint32toBytes(sizeBuf, 4)
		buf.Write(sizeBuf)
		buf.Write([]byte{0xff, 0xff, 0xff, 0xff})
		if _, _, err := runLoadEntries(c, &buf, client); err == nil {
			t.Fatal("loadEntries returned nil on malformed proto")
		}
	})
}

func writeCorruptTail(buf *bytes.Buffer) {
	sizeBuf := make([]byte, 4)
	util.Uint32toBytes(sizeBuf, 100)
	buf.Write(sizeBuf)
	buf.Write([]byte{1, 2, 3})
}