seaweedfs/weed/shell/command_fs_merge_volumes_test.go

package shell

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"sync"
	"testing"

	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/util"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

func newMergeCmd(limitMB uint64, vols ...*master_pb.VolumeInformationMessage) *commandFsMergeVolumes {
	c := &commandFsMergeVolumes{
		volumes:         make(map[needle.VolumeId]*master_pb.VolumeInformationMessage),
		volumeSizeLimit: limitMB * 1024 * 1024,
	}
	for _, v := range vols {
		c.volumes[needle.VolumeId(v.Id)] = v
	}
	return c
}

// An explicit -fromVolumeId/-toVolumeId pair must be honored as given, even when
// the source is larger than the target. The heuristic planner only ever merges
// a smaller volume into a larger one, which used to make this request a silent
// no-op.
func TestCreateMergePlan_HonorsExplicitDirection(t *testing.T) {
	larger := &master_pb.VolumeInformationMessage{Id: 87, Size: 7192590976}
	smaller := &master_pb.VolumeInformationMessage{Id: 83, Size: 7088822248}
	c := newMergeCmd(250000, larger, smaller)

	plan, err := c.createMergePlan("*", []needle.VolumeId{83}, needle.VolumeId(87))
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got := plan.targets[needle.VolumeId(87)]; len(got) != 1 || got[0] != needle.VolumeId(83) {
		t.Fatalf("expected 87->83, got plan=%v", plan.targets)
	}

	// The reverse direction keeps working too.
	plan, err = c.createMergePlan("*", []needle.VolumeId{87}, needle.VolumeId(83))
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got := plan.targets[needle.VolumeId(83)]; len(got) != 1 || got[0] != needle.VolumeId(87) {
		t.Fatalf("expected 83->87, got plan=%v", plan.targets)
	}
}

// Merging into an empty target is valid (e.g. consolidating into a freshly
// vacuumed volume); only an empty source should be rejected.
func TestCreateMergePlan_DirectedAllowsEmptyTarget(t *testing.T) {
	from := &master_pb.VolumeInformationMessage{Id: 87, Size: 100}
	emptyTo := &master_pb.VolumeInformationMessage{Id: 83, Size: 0}
	c := newMergeCmd(250000, from, emptyTo)

	plan, err := c.createMergePlan("*", []needle.VolumeId{83}, needle.VolumeId(87))
	if err != nil {
		t.Fatalf("unexpected error merging into empty target: %v", err)
	}
	if got := plan.targets[needle.VolumeId(87)]; len(got) != 1 || got[0] != needle.VolumeId(83) {
		t.Fatalf("expected 87->83, got plan=%v", plan.targets)
	}
}

// The directed planner rejects a self-map on its own, not just via Do, since it
// is exercised directly.
func TestCreateMergePlan_DirectedRejectsSelfMap(t *testing.T) {
	v := &master_pb.VolumeInformationMessage{Id: 87, Size: 100}
	c := newMergeCmd(250000, v)

	_, err := c.createMergePlan("*", []needle.VolumeId{87}, needle.VolumeId(87))
	if err == nil || !strings.Contains(err.Error(), "no volume id changes") {
		t.Fatalf("expected self-map rejection, got %v", err)
	}
}

func TestCreateMergePlan_DirectedRejectsIneligible(t *testing.T) {
	cases := []struct {
		name     string
		from, to *master_pb.VolumeInformationMessage
		coll     string
		wantErr  string
	}{
		{
			name:    "readonly source",
			from:    &master_pb.VolumeInformationMessage{Id: 87, Size: 100, ReadOnly: true},
			to:      &master_pb.VolumeInformationMessage{Id: 83, Size: 100},
			coll:    "*",
			wantErr: "volume 87 is readonly",
		},
		{
			name:    "readonly target",
			from:    &master_pb.VolumeInformationMessage{Id: 87, Size: 100},
			to:      &master_pb.VolumeInformationMessage{Id: 83, Size: 100, ReadOnly: true},
			coll:    "*",
			wantErr: "volume 83 is readonly",
		},
		{
			name:    "empty source",
			from:    &master_pb.VolumeInformationMessage{Id: 87, Size: 0},
			to:      &master_pb.VolumeInformationMessage{Id: 83, Size: 100},
			coll:    "*",
			wantErr: "volume 87 is empty",
		},
		{
			name:    "wrong collection",
			from:    &master_pb.VolumeInformationMessage{Id: 87, Size: 100, Collection: "other"},
			to:      &master_pb.VolumeInformationMessage{Id: 83, Size: 100, Collection: "other"},
			coll:    "wanted",
			wantErr: "volume 87 does not match collection filter",
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			c := newMergeCmd(250000, tc.from, tc.to)
			_, err := c.createMergePlan(tc.coll, []needle.VolumeId{needle.VolumeId(tc.to.Id)}, needle.VolumeId(tc.from.Id))
			if err == nil {
				t.Fatalf("expected error %q, got nil", tc.wantErr)
			}
			if got := err.Error(); !strings.Contains(got, tc.wantErr) {
				t.Fatalf("expected error containing %q, got %q", tc.wantErr, got)
			}
		})
	}
}

// A source that fits into no single target distributes across several, each
// chunk going to the target with the most remaining capacity.
func TestCreateMergePlan_DistributesAcrossMultipleTargets(t *testing.T) {
	mb := uint64(1024 * 1024)
	from := &master_pb.VolumeInformationMessage{Id: 112, Size: 60 * mb}
	to1 := &master_pb.VolumeInformationMessage{Id: 111, Size: 60 * mb}
	to2 := &master_pb.VolumeInformationMessage{Id: 107, Size: 70 * mb}
	c := newMergeCmd(100, from, to1, to2)

	plan, err := c.createMergePlan("*", []needle.VolumeId{111, 107}, needle.VolumeId(112))
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got := plan.targets[needle.VolumeId(112)]; len(got) != 2 {
		t.Fatalf("expected two targets, got %v", got)
	}

	// 111 has 40 MB free, 107 has 30 MB free. The first chunk goes to the
	// emptier 111; allocations keep balancing remaining capacity after that.
	vid, ok := plan.allocate(needle.VolumeId(112), 5*mb)
	if !ok || vid != needle.VolumeId(111) {
		t.Fatalf("expected first chunk on 111, got %v ok=%v", vid, ok)
	}
	vid, ok = plan.allocate(needle.VolumeId(112), 10*mb)
	if !ok || vid != needle.VolumeId(111) {
		t.Fatalf("expected second chunk on 111 (35 MB free vs 30), got %v ok=%v", vid, ok)
	}
	vid, ok = plan.allocate(needle.VolumeId(112), 5*mb)
	if !ok || vid != needle.VolumeId(107) {
		t.Fatalf("expected third chunk on 107 (30 MB free vs 25), got %v ok=%v", vid, ok)
	}
	// A chunk larger than any remaining capacity is refused.
	if vid, ok = plan.allocate(needle.VolumeId(112), 45*mb); ok {
		t.Fatalf("expected no room for oversized chunk, got %v", vid)
	}
	// Smaller chunks still fit afterwards.
	if _, ok = plan.allocate(needle.VolumeId(112), 20*mb); !ok {
		t.Fatal("expected room for 20 MB chunk")
	}

	// A failed move's release makes the reservation reusable.
	vid2, ok := plan.allocate(needle.VolumeId(112), 20*mb)
	if !ok {
		t.Fatal("expected room for second 20 MB chunk")
	}
	plan.release(needle.VolumeId(112), vid2, 20*mb)
	if vid3, ok := plan.allocate(needle.VolumeId(112), 20*mb); !ok || vid3 != vid2 {
		t.Fatalf("released capacity must be reusable, got %v ok=%v", vid3, ok)
	}
}

// The combined free capacity of all targets must cover the source's live data.
func TestCreateMergePlan_MultiTargetRejectsInsufficientCapacity(t *testing.T) {
	mb := uint64(1024 * 1024)
	from := &master_pb.VolumeInformationMessage{Id: 112, Size: 90 * mb}
	to1 := &master_pb.VolumeInformationMessage{Id: 111, Size: 60 * mb}
	to2 := &master_pb.VolumeInformationMessage{Id: 107, Size: 60 * mb}
	c := newMergeCmd(100, from, to1, to2)

	_, err := c.createMergePlan("*", []needle.VolumeId{111, 107}, needle.VolumeId(112))
	if err == nil || !strings.Contains(err.Error(), "cannot merge into volumes") {
		t.Fatalf("expected capacity rejection, got %v", err)
	}
}

// Without -fromVolumeId, a -toVolumeId list restricts the heuristic planner's
// target candidates to the listed volumes.
func TestCreateMergePlan_TargetListRestrictsHeuristic(t *testing.T) {
	mb := uint64(1024 * 1024)
	big := &master_pb.VolumeInformationMessage{Id: 1, Size: 80 * mb}
	mid := &master_pb.VolumeInformationMessage{Id: 2, Size: 50 * mb}
	small := &master_pb.VolumeInformationMessage{Id: 3, Size: 10 * mb}
	c := newMergeCmd(100, big, mid, small)

	plan, err := c.createMergePlan("*", []needle.VolumeId{2}, 0)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got := plan.targets[needle.VolumeId(3)]; len(got) != 1 || got[0] != needle.VolumeId(2) {
		t.Fatalf("expected 3->2, got %v", plan.targets)
	}
	if plan.isSource(needle.VolumeId(1)) {
		t.Fatalf("volume 1 must not merge anywhere, got %v", plan.targets)
	}
}

func TestParseTargetVolumeIds(t *testing.T) {
	if ids, err := parseTargetVolumeIds(""); err != nil || ids != nil {
		t.Fatalf("empty arg: got %v, %v", ids, err)
	}
	if ids, err := parseTargetVolumeIds("0"); err != nil || ids != nil {
		t.Fatalf("zero arg: got %v, %v", ids, err)
	}
	ids, err := parseTargetVolumeIds("111, 107")
	if err != nil || len(ids) != 2 || ids[0] != needle.VolumeId(111) || ids[1] != needle.VolumeId(107) {
		t.Fatalf("list arg: got %v, %v", ids, err)
	}
	if _, err = parseTargetVolumeIds("111,111"); err == nil {
		t.Fatal("expected duplicate rejection")
	}
	if _, err = parseTargetVolumeIds("111,x"); err == nil {
		t.Fatal("expected invalid id rejection")
	}
	if _, err = parseTargetVolumeIds("4294967297"); err == nil {
		t.Fatal("expected uint32 overflow rejection")
	}
}

// The collection filter speaks the shared vocabulary: "_default" selects the
// empty-named collection, a name or pattern only its matches, and "*" (or no
// flag) every collection.
func TestCreateMergePlan_CollectionFilter(t *testing.T) {
	mb := uint64(1024 * 1024)
	mk := func(id uint32, sizeMB uint64, collection string) *master_pb.VolumeInformationMessage {
		return &master_pb.VolumeInformationMessage{Id: id, Size: sizeMB * mb, Collection: collection}
	}
	newCmd := func() *commandFsMergeVolumes {
		return newMergeCmd(250000, mk(1, 10, ""), mk(2, 80, ""), mk(3, 10, "x"), mk(4, 80, "x"))
	}

	// "_default" merges only the unnamed-collection volumes.
	plan, err := newCmd().createMergePlan(CollectionDefault, nil, 0)
	if err != nil {
		t.Fatalf("%s: %v", CollectionDefault, err)
	}
	if got := plan.targets[needle.VolumeId(1)]; len(got) != 1 || got[0] != needle.VolumeId(2) {
		t.Fatalf("expected 1->2, got %v", plan.targets)
	}
	if plan.isSource(needle.VolumeId(3)) {
		t.Fatalf("named-collection volume must not merge under _default, got %v", plan.targets)
	}

	// A literal name still selects its collection.
	plan, err = newCmd().createMergePlan("x", nil, 0)
	if err != nil {
		t.Fatalf("x: %v", err)
	}
	if got := plan.targets[needle.VolumeId(3)]; len(got) != 1 || got[0] != needle.VolumeId(4) {
		t.Fatalf("expected 3->4, got %v", plan.targets)
	}
	if plan.isSource(needle.VolumeId(1)) {
		t.Fatalf("unnamed volume must not merge under x, got %v", plan.targets)
	}

	// "*" keeps covering every collection.
	plan, err = newCmd().createMergePlan("*", nil, 0)
	if err != nil {
		t.Fatalf("*: %v", err)
	}
	for src, dest := range map[needle.VolumeId]needle.VolumeId{1: 2, 3: 4} {
		if got := plan.targets[src]; len(got) != 1 || got[0] != dest {
			t.Fatalf("expected %d->%d, got %v", src, dest, plan.targets)
		}
	}
}

// A volume reporting more deleted bytes than it holds must read as empty.
func TestGetVolumeSize_ClampsDeletedOverSize(t *testing.T) {
	c := newMergeCmd(250000)

	overDeleted := &master_pb.VolumeInformationMessage{Id: 1, Size: 1000, DeletedByteCount: 4000}
	if got := c.getVolumeSize(overDeleted); got != 0 {
		t.Errorf("expected 0 for a volume with more deleted than stored, got %d", got)
	}

	healthy := &master_pb.VolumeInformationMessage{Id: 2, Size: 3000, DeletedByteCount: 1000}
	if got := c.getVolumeSize(healthy); got != 2000 {
		t.Errorf("expected 2000, got %d", got)
	}
}

// Manifests on volumes in a foreign collection cannot reference plan volumes
// and must be skipped instead of downloaded; plan sources and unknown volumes
// must still be resolved.
func TestManifestMayReferencePlan(t *testing.T) {
	src := &master_pb.VolumeInformationMessage{Id: 1, Size: 100, Collection: "x"}
	same := &master_pb.VolumeInformationMessage{Id: 2, Size: 100, Collection: "x"}
	foreign := &master_pb.VolumeInformationMessage{Id: 3, Size: 100, Collection: "y"}
	c := newMergeCmd(250000, src, same, foreign)

	plan := newMergePlan(c.volumeSizeLimit)
	plan.targets[needle.VolumeId(1)] = []needle.VolumeId{2}
	planCollections := map[string]bool{"x": true}

	for vid, want := range map[needle.VolumeId]bool{
		1: true,  // plan source
		2: true,  // same collection
		3: false, // foreign collection
		9: true,  // unknown volume, resolve conservatively
	} {
		if got := c.manifestMayReferencePlan(plan, planCollections, vid); got != want {
			t.Fatalf("volume %d: got %v, want %v", vid, got, want)
		}
	}
}

// A source whose index still holds needles but that the traversal never saw is
// made entirely of orphan needles — the merge moves nothing, but the operator
// must be told the real cleanup is fsck.
func TestWarnUnreferencedSources(t *testing.T) {
	vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 18}
	c := newMergeCmd(250000, vol)
	plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
		needle.VolumeId(203): {needle.VolumeId(187)},
	}}

	var sb strings.Builder
	c.warnUnreferencedSources(&sb, plan, newSourceNeedleCounter(), "/buckets/test")
	out := sb.String()
	if !strings.Contains(out, "volume 203") || !strings.Contains(out, "orphan needles") {
		t.Fatalf("expected orphan warning, got %q", out)
	}

	// Needles seen during traversal: no warning.
	seen := newSourceNeedleCounter()
	for i := 0; i < 7; i++ {
		seen.record(needle.VolumeId(203))
	}
	sb.Reset()
	c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test")
	if sb.Len() != 0 {
		t.Fatalf("expected no warning when needles were seen, got %q", sb.String())
	}

	// Zero FileCount (truly empty volume): nothing to report.
	emptyVol := &master_pb.VolumeInformationMessage{Id: 204}
	c2 := newMergeCmd(250000, emptyVol)
	plan2 := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
		needle.VolumeId(204): {needle.VolumeId(187)},
	}}
	sb.Reset()
	c2.warnUnreferencedSources(&sb, plan2, newSourceNeedleCounter(), "/buckets/test")
	if sb.Len() != 0 {
		t.Fatalf("expected no warning for empty volume, got %q", sb.String())
	}
}

// Concurrent BFS workers increment the production counter through record — the
// count must stay correct under -race with source-heavy inputs.
func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) {
	vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 1000}
	c := newMergeCmd(250000, vol)
	plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
		needle.VolumeId(203): {needle.VolumeId(187)},
	}}

	seen := newSourceNeedleCounter()
	var wg sync.WaitGroup
	for w := 0; w < 5; w++ { // TraverseBfs worker count
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 200; i++ {
				seen.record(needle.VolumeId(203))
			}
		}()
	}
	wg.Wait()

	if got := seen.count(needle.VolumeId(203)); got != 1000 {
		t.Fatalf("lost increments under concurrency: %d", got)
	}

	var sb strings.Builder
	c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test")
	if sb.Len() != 0 {
		t.Fatalf("expected no warning when all needles were seen, got %q", sb.String())
	}
}

type conditionalUpdateFiler struct {
	filer_pb.SeaweedFilerClient
	err       error
	req       *filer_pb.UpdateEntryRequest
	current   *filer_pb.Entry
	lookupErr error
	lookupReq *filer_pb.LookupDirectoryEntryRequest
}

func (s *conditionalUpdateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
	s.lookupReq = req
	if s.lookupErr != nil {
		return nil, s.lookupErr
	}
	return &filer_pb.LookupDirectoryEntryResponse{Entry: s.current}, nil
}

func (s *conditionalUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
	s.req = req
	if s.err != nil {
		return nil, s.err
	}
	return &filer_pb.UpdateEntryResponse{}, nil
}

// The write-back is conditioned on the entry as it was read, and only a refusal
// that proves nothing was written counts as stale: the caller then deletes the
// copies it made, which would lose data if the update might have landed.
func TestUpdateEntryIfUnchanged(t *testing.T) {
	read := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}}
	moved := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 7}}}}

	for _, tc := range []struct {
		name  string
		err   error
		stale bool
	}{
		{"committed", nil, false},
		{"changed", status.Error(codes.FailedPrecondition, "precondition failed"), true},
		{"deleted", status.Error(codes.NotFound, "not found"), true},
		{"outcome unknown", status.Error(codes.Unavailable, "filer down"), false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			fake := &conditionalUpdateFiler{err: tc.err}
			stale, err := updateEntryIfUnchanged(context.Background(), fake, "/dir", moved, read)
			if stale != tc.stale || (err != nil) != (tc.err != nil) {
				t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err)
			}
			clauses := fake.req.GetCondition().GetClauses()
			if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL ||
				!proto.Equal(clauses[0].ExpectedEntry, read) || !proto.Equal(fake.req.Entry, moved) {
				t.Fatalf("request %v must write the moved entry conditioned on the read one", fake.req)
			}
		})
	}
}

// A refused update may have been refused by an overlapping run that moved the
// same chunk to the same volume: moveChunk keeps the key and cookie, so that
// run's committed entry points at the very fid this run would delete.
func TestUnreferencedTargets(t *testing.T) {
	committed := orphanedNeedle{volumeId: 9, fileId: "9,0101000007"}
	own := orphanedNeedle{volumeId: 9, fileId: "9,0202000007"}
	manifestCopy := orphanedNeedle{volumeId: 9, fileId: "9,0303000007"}
	targets := func() []orphanedNeedle { return []orphanedNeedle{committed, own, manifestCopy} }
	direct := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId}}}

	blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId, Size: 10}}})
	if err != nil {
		t.Fatal(err)
	}
	volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write(blob)
	}))
	defer volumeServer.Close()
	lookupFn := func(_ context.Context, fileId string) ([]string, error) {
		if fileId != manifestCopy.fileId {
			return nil, fmt.Errorf("volume of %s unavailable", fileId)
		}
		return []string{volumeServer.URL + "/" + fileId}, nil
	}
	viaManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: manifestCopy.fileId, IsChunkManifest: true, Size: 10}}}
	unreadableManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: "9,0404000007", IsChunkManifest: true, Size: 10}}}

	for _, tc := range []struct {
		name      string
		current   *filer_pb.Entry
		lookupErr error
		want      []orphanedNeedle
	}{
		{"another run committed one copy", direct, nil, []orphanedNeedle{own, manifestCopy}},
		{"another run committed copies through a manifest", viaManifest, nil, []orphanedNeedle{own}},
		{"manifest unreadable", unreadableManifest, nil, nil},
		{"entry deleted", direct, status.Error(codes.NotFound, "not found"), targets()},
		{"entry unreadable", direct, status.Error(codes.Unavailable, "filer down"), nil},
	} {
		t.Run(tc.name, func(t *testing.T) {
			fake := &conditionalUpdateFiler{current: tc.current, lookupErr: tc.lookupErr}
			got := unreferencedTargets(context.Background(), fake, lookupFn, "/dir", "f", targets())
			if !slices.Equal(got, tc.want) {
				t.Fatalf("got %v, want %v", got, tc.want)
			}
			if fake.lookupReq.GetDirectory() != "/dir" || fake.lookupReq.GetName() != "f" {
				t.Fatalf("looked up %v, want /dir f", fake.lookupReq)
			}
		})
	}
}

func TestIsMultipartUploadFile(t *testing.T) {
	cases := []struct {
		buckets, dir string
		want         bool
	}{
		{"/buckets", "/buckets/b/.uploads/abc", true},
		{"/", "/b/.uploads/abc", true},
		{"/", "/buckets/b/.uploads/abc", false},
		{"/buckets/", "/buckets/b/.uploads/abc", true},
		{"/data/buckets", "/data/buckets/b/.uploads/abc", true},
		{"/buckets", "/buckets/b/.uploads", false},
		{"/buckets", "/buckets/b/dir/.uploads/abc", false},
		{"/buckets", "/buckets/b/.uploads/abc/sub", false},
		{"/buckets", "/other/b/.uploads/abc", false},
		{"/buckets", "/buckets/b/obj", false},
	}
	for _, tc := range cases {
		if got := isMultipartUploadFile(tc.buckets, util.FullPath(tc.dir)); got != tc.want {
			t.Errorf("isMultipartUploadFile(%q, %q) = %v, want %v", tc.buckets, tc.dir, got, tc.want)
		}
	}
}

func TestRecordPlanSources(t *testing.T) {
	plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
		needle.VolumeId(203): {needle.VolumeId(187)},
	}}
	c := newMergeCmd(250000,
		&master_pb.VolumeInformationMessage{Id: 203, Collection: "b"},
		&master_pb.VolumeInformationMessage{Id: 187, Collection: "b"},
		&master_pb.VolumeInformationMessage{Id: 5, Collection: "other"})
	planCollections := map[string]bool{"b": true}
	chunk := func(vid uint32, key uint64) *filer_pb.FileChunk {
		fid := &filer_pb.FileId{VolumeId: vid, FileKey: key, Cookie: 7}
		return &filer_pb.FileChunk{Fid: fid, FileId: needle.NewFileId(needle.VolumeId(vid), key, 7).String(), Size: 10}
	}

	blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{chunk(203, 2)}})
	if err != nil {
		t.Fatal(err)
	}
	volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write(blob)
	}))
	defer volumeServer.Close()
	lookupFn := func(_ context.Context, fileId string) ([]string, error) {
		return []string{volumeServer.URL + "/" + fileId}, nil
	}
	manifest := func(vid uint32) *filer_pb.FileChunk {
		m := chunk(vid, 1)
		m.IsChunkManifest = true
		return m
	}

	for _, tc := range []struct {
		name   string
		chunks []*filer_pb.FileChunk
		found  bool
		count  int
	}{
		{"off plan sources", []*filer_pb.FileChunk{chunk(187, 1), chunk(5, 1)}, false, 0},
		{"on a plan source", []*filer_pb.FileChunk{chunk(203, 1), chunk(187, 2), chunk(203, 3)}, true, 2},
		{"behind a manifest off the source", []*filer_pb.FileChunk{manifest(187)}, true, 1},
		{"manifest outside the plan collections", []*filer_pb.FileChunk{manifest(5)}, false, 0},
	} {
		t.Run(tc.name, func(t *testing.T) {
			seen := newSourceNeedleCounter()
			found, err := c.recordPlanSources(context.Background(), lookupFn, plan, planCollections, seen, tc.chunks)
			if err != nil {
				t.Fatal(err)
			}
			if found != tc.found || seen.count(needle.VolumeId(203)) != tc.count {
				t.Fatalf("found %v with %d references to 203, want %v with %d", found, seen.count(needle.VolumeId(203)), tc.found, tc.count)
			}
		})
	}

	unreachable := func(_ context.Context, fileId string) ([]string, error) {
		return nil, fmt.Errorf("volume of %s unavailable", fileId)
	}
	seen := newSourceNeedleCounter()
	found, err := c.recordPlanSources(context.Background(), unreachable, plan, planCollections, seen, []*filer_pb.FileChunk{chunk(203, 1), manifest(187)})
	if err == nil {
		t.Fatal("an unreadable manifest must be reported")
	}
	if !found || seen.count(needle.VolumeId(203)) != 1 {
		t.Fatalf("direct reference lost with an unreadable manifest: found %v, %d references to 203", found, seen.count(needle.VolumeId(203)))
	}

	unreadable := chunk(187, 10)
	unreadable.IsChunkManifest = true
	nestedBlob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{unreadable}})
	if err != nil {
		t.Fatal(err)
	}
	outer := chunk(187, 9)
	outer.IsChunkManifest = true
	partServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path == "/"+outer.GetFileIdString() {
			_, _ = w.Write(nestedBlob)
			return
		}
		_, _ = w.Write(blob)
	}))
	defer partServer.Close()
	failsLater := func(_ context.Context, fileId string) ([]string, error) {
		if fileId == unreadable.GetFileIdString() {
			return nil, fmt.Errorf("volume of %s unavailable", fileId)
		}
		return []string{partServer.URL + "/" + fileId}, nil
	}
	seen = newSourceNeedleCounter()
	found, err = c.recordPlanSources(context.Background(), failsLater, plan, planCollections, seen, []*filer_pb.FileChunk{manifest(187), outer})
	if err == nil {
		t.Fatal("an unreadable manifest must be reported")
	}
	if !found || seen.count(needle.VolumeId(203)) != 1 {
		t.Fatalf("chunks behind a readable manifest lost when a later one is unreadable: found %v, %d references to 203", found, seen.count(needle.VolumeId(203)))
	}
}