volume: read a manifest chunk from another replica when one fails (#12313)

* operation: read a manifest chunk from the next replica when one fails ChunkedFileReader read each chunk from one random location and returned the first error, so a chunk whose cached replica went down failed the whole file although another replica had it. Try the chunk's locations in turn (local-data replicas first) and resume each attempt at the byte the previous one reached. A failed write to the caller still ends the read. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * operation: drop a chunk volume's cached locations after a failed read A chunk read that failed on a replica left that replica in the 10-minute location cache, so later reads kept trying it first. Drop the volume's cached locations after a failure, and when every cached replica has failed, ask the master once more and try the locations it now lists. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * operation: don't continue a partly sent chunk on another replica A volume server streams a large needle before it has verified the checksum and aborts the response on a mismatch, so bytes that already reached the reader may be corrupt. Continuing such a chunk on another replica would hand back a mix of bad and good bytes as a complete read. Move to the next replica only while nothing of the chunk has been written; a mid-body failure still fails the read, and the dropped cache entry lets the next read pick a live replica. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * operation: treat a short chunk body as a failed replica read A replica that ends a chunk's body cleanly before the chunk's size made the read look successful, so the next replica was never tried and the following chunks landed at the wrong offsets. Read exactly the chunk's remaining length and report a short body as an error. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 10:17:10 +01:00 · 2026-11-01 16:31:22 -08:00
1 parent ab204a03e6
commit 5cf053c32e
@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"sort"
+	"strings"
 	"sync"

 	"google.golang.org/grpc"
@@ -94,7 +95,7 @@ func (cm *ChunkManifest) DeleteChunks(masterFn GetMasterFn, usePublicUrl bool, g
 	return nil
 }

-func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (written int64, e error) {
+func readChunkNeedle(fileUrl string, w io.Writer, offset int64, size int64, jwt string) (written int64, e error) {
 	req, err := http.NewRequest(http.MethodGet, fileUrl, nil)
 	if err != nil {
 		return written, err
@@ -125,7 +126,11 @@ func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (wri
 		return written, util_http.StatusError(fileUrl, resp)

 	}
-	return io.Copy(w, resp.Body)
+	written, e = io.CopyN(w, resp.Body, size)
+	if e == io.EOF {
+		e = fmt.Errorf("%s: %d of %d bytes: %w", fileUrl, written, size, io.ErrUnexpectedEOF)
+	}
+	return written, e
 }

 func NewChunkedFileReader(chunkList []*ChunkInfo, master pb.ServerAddress, grpcDialOption grpc.DialOption) *ChunkedFileReader {
@@ -175,26 +180,69 @@ func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) {
 		return n, ErrInvalidRange
 	}
 	for ; chunkIndex < len(cf.chunkList); chunkIndex++ {
-		ci := cf.chunkList[chunkIndex]
-		// if we need read date from local volume server first?
-		fileUrl, jwt, lookupError := LookupFileId(func(_ context.Context) pb.ServerAddress {
-			return cf.master
-		}, cf.grpcDialOption, ci.Fid)
-		if lookupError != nil {
-			return n, lookupError
-		}
-		if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset, jwt); e != nil {
+		wn, e := cf.readChunk(cf.chunkList[chunkIndex], w, chunkStartOffset)
+		n += wn
+		cf.pos += wn
+		if e != nil {
 			return n, e
-		} else {
-			n += wn
-			cf.pos += wn
 		}
-
 		chunkStartOffset = 0
 	}
 	return n, nil
 }

+// readChunk reads the chunk from the first replica that serves it. A failure
+// drops the volume's cached locations, and once every cached replica has
+// failed the master is asked again. A replica that already sent part of the
+// chunk is not continued elsewhere: a volume server streams a needle before
+// its checksum is verified, so those bytes may be the corrupt ones.
+func (cf *ChunkedFileReader) readChunk(ci *ChunkInfo, w io.Writer, offset int64) (written int64, err error) {
+	masterFn := func(_ context.Context) pb.ServerAddress {
+		return cf.master
+	}
+	vid, _, _ := strings.Cut(ci.Fid, ",")
+	dst := &writeErrorRecorder{Writer: w}
+	tried := make(map[string]bool)
+	for range 2 {
+		fileUrls, jwt, lookupError := lookupFileIdUrls(masterFn, cf.grpcDialOption, ci.Fid)
+		if lookupError != nil {
+			if err == nil {
+				err = lookupError
+			}
+			return 0, err
+		}
+		for _, fileUrl := range fileUrls {
+			if tried[fileUrl] {
+				continue
+			}
+			tried[fileUrl] = true
+			written, err = readChunkNeedle(fileUrl, dst, offset, ci.Size-offset, jwt)
+			if err == nil || dst.err != nil {
+				return written, err
+			}
+			glog.V(0).Infof("read chunk %s from %s: %v", ci.Fid, fileUrl, err)
+			InvalidateVolumeIdLocationCache(vid)
+			if written > 0 {
+				return written, err
+			}
+		}
+	}
+	return 0, err
+}
+
+type writeErrorRecorder struct {
+	io.Writer
+	err error
+}
+
+func (r *writeErrorRecorder) Write(p []byte) (int, error) {
+	n, err := r.Writer.Write(p)
+	if err != nil {
+		r.err = err
+	}
+	return n, err
+}
+
 func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) {
 	cf.Seek(off, 0)
 	return cf.Read(p)
@@ -0,0 +1,257 @@
+package operation
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type fakeVolumeLocationServer struct {
+	master_pb.UnimplementedSeaweedServer
+	mu    sync.Mutex
+	calls int
+	urls  []string
+}
+
+func (s *fakeVolumeLocationServer) LookupVolume(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls++
+	resp := &master_pb.LookupVolumeResponse{}
+	for _, vid := range req.VolumeOrFileIds {
+		var locs []*master_pb.Location
+		for _, u := range s.urls {
+			locs = append(locs, &master_pb.Location{Url: u})
+		}
+		resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{
+			VolumeOrFileId: vid,
+			Locations:      locs,
+		})
+	}
+	return resp, nil
+}
+
+func (s *fakeVolumeLocationServer) lookups() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.calls
+}
+
+type chunkReplica struct {
+	*httptest.Server
+	mu     sync.Mutex
+	ranges []string
+}
+
+func (c *chunkReplica) host() string {
+	return strings.TrimPrefix(c.URL, "http://")
+}
+
+func (c *chunkReplica) requests() []string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]string(nil), c.ranges...)
+}
+
+// newChunkReplica serves data; with cutAfter >= 0 it announces the full length
+// but drops the connection after cutAfter bytes.
+func newChunkReplica(t *testing.T, data []byte, cutAfter int) *chunkReplica {
+	c := &chunkReplica{}
+	c.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c.mu.Lock()
+		c.ranges = append(c.ranges, r.Header.Get("Range"))
+		c.mu.Unlock()
+		if cutAfter < 0 {
+			http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(data))
+			return
+		}
+		w.Header().Set("Content-Length", strconv.Itoa(len(data)))
+		w.WriteHeader(http.StatusOK)
+		w.Write(data[:cutAfter])
+		w.(http.Flusher).Flush()
+		conn, _, err := w.(http.Hijacker).Hijack()
+		if err == nil {
+			conn.Close()
+		}
+	}))
+	t.Cleanup(c.Close)
+	return c
+}
+
+func deadAddress(t *testing.T) string {
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	addr := lis.Addr().String()
+	lis.Close()
+	return addr
+}
+
+func testChunkData(size int) []byte {
+	data := make([]byte, size)
+	for i := range data {
+		data[i] = byte(i * 7)
+	}
+	return data
+}
+
+func TestChunkedFileReaderTriesTheNextReplica(t *testing.T) {
+	const vid = "91201"
+	data := testChunkData(4096)
+	healthy := newChunkReplica(t, data, -1)
+	fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}}
+	master := startFakeMasterServer(t, fake)
+	dial := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	vc.Set(vid, []Location{{Url: deadAddress(t)}, {Url: healthy.host(), DataInRemote: true}}, time.Minute)
+	t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) })
+
+	chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}}
+	reader := NewChunkedFileReader(chunks, master, dial)
+	reader.Seek(1000, io.SeekStart)
+	got, err := io.ReadAll(reader)
+	reader.Close()
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if !bytes.Equal(got, data[1000:]) {
+		t.Fatalf("read %d bytes, want the last %d bytes of the chunk", len(got), len(data)-1000)
+	}
+	if r := healthy.requests(); len(r) != 1 || r[0] != "bytes=1000-" {
+		t.Fatalf("healthy replica requests = %q, want one from byte 1000", r)
+	}
+	if _, err := vc.Get(vid); err == nil {
+		t.Fatalf("cached locations kept after a replica failed")
+	}
+	if fake.lookups() != 0 {
+		t.Fatalf("master lookups = %d, want 0", fake.lookups())
+	}
+}
+
+func TestChunkedFileReaderDoesNotContinueAPartlySentChunk(t *testing.T) {
+	const vid = "91205"
+	data := testChunkData(4096)
+	broken := newChunkReplica(t, data, 1000)
+	healthy := newChunkReplica(t, data, -1)
+	fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}}
+	master := startFakeMasterServer(t, fake)
+	dial := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	vc.Set(vid, []Location{{Url: broken.host()}, {Url: healthy.host(), DataInRemote: true}}, time.Minute)
+	t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) })
+
+	chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}}
+	reader := NewChunkedFileReader(chunks, master, dial)
+	_, err := io.ReadAll(reader)
+	reader.Close()
+	if err == nil {
+		t.Fatalf("read of a chunk cut off mid-body succeeded")
+	}
+	if r := healthy.requests(); len(r) != 0 {
+		t.Fatalf("healthy replica requests = %q, want none", r)
+	}
+
+	reader = NewChunkedFileReader(chunks, master, dial)
+	got, err := io.ReadAll(reader)
+	reader.Close()
+	if err != nil || !bytes.Equal(got, data) {
+		t.Fatalf("second read: %d bytes, err %v", len(got), err)
+	}
+	if fake.lookups() != 1 || len(broken.requests()) != 1 {
+		t.Fatalf("second read: %d master lookups, %d requests to the broken replica; want 1 and 1", fake.lookups(), len(broken.requests()))
+	}
+}
+
+func TestChunkedFileReaderAsksMasterWhenCachedReplicasFail(t *testing.T) {
+	const vid = "91202"
+	first, second := testChunkData(3000), testChunkData(2000)
+	healthy := newChunkReplica(t, second, -1)
+	fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}}
+	master := startFakeMasterServer(t, fake)
+	dial := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	firstReplica := newChunkReplica(t, first, -1)
+	const otherVid = "91203"
+	vc.Set(otherVid, []Location{{Url: firstReplica.host()}}, time.Minute)
+	vc.Set(vid, []Location{{Url: deadAddress(t)}}, time.Minute)
+	t.Cleanup(func() {
+		InvalidateVolumeIdLocationCache(vid)
+		InvalidateVolumeIdLocationCache(otherVid)
+	})
+
+	chunks := []*ChunkInfo{
+		{Fid: otherVid + ",01637037d6", Offset: 0, Size: int64(len(first))},
+		{Fid: vid + ",02637037d6", Offset: int64(len(first)), Size: int64(len(second))},
+	}
+	reader := NewChunkedFileReader(chunks, master, dial)
+	defer reader.Close()
+	got, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if want := append(append([]byte(nil), first...), second...); !bytes.Equal(got, want) {
+		t.Fatalf("read %d bytes, want %d", len(got), len(want))
+	}
+	if fake.lookups() != 1 {
+		t.Fatalf("master lookups = %d, want 1", fake.lookups())
+	}
+}
+
+func TestChunkedFileReaderTreatsAShortBodyAsFailed(t *testing.T) {
+	const vid = "91206"
+	data := testChunkData(2048)
+	empty := newChunkReplica(t, nil, -1)
+	healthy := newChunkReplica(t, data, -1)
+	vc.Set(vid, []Location{{Url: empty.host()}, {Url: healthy.host(), DataInRemote: true}}, time.Minute)
+	t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) })
+
+	chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}}
+	reader := NewChunkedFileReader(chunks, pb.ServerAddress("127.0.0.1:1"), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	defer reader.Close()
+	got, err := io.ReadAll(reader)
+	if err != nil || !bytes.Equal(got, data) {
+		t.Fatalf("read: %d bytes, err %v; want %d bytes", len(got), err, len(data))
+	}
+	if len(empty.requests()) != 1 || len(healthy.requests()) != 1 {
+		t.Fatalf("requests: empty %d, healthy %d; want 1 and 1", len(empty.requests()), len(healthy.requests()))
+	}
+}
+
+type failingWriter struct{}
+
+func (failingWriter) Write([]byte) (int, error) { return 0, errors.New("client gone") }
+
+func TestChunkedFileReaderStopsWhenTheClientGoesAway(t *testing.T) {
+	const vid = "91204"
+	data := testChunkData(1024)
+	a, b := newChunkReplica(t, data, -1), newChunkReplica(t, data, -1)
+	vc.Set(vid, []Location{{Url: a.host()}, {Url: b.host()}}, time.Minute)
+	t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) })
+
+	chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}}
+	reader := NewChunkedFileReader(chunks, pb.ServerAddress("127.0.0.1:1"), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if _, err := reader.WriteTo(failingWriter{}); err == nil || err.Error() != "client gone" {
+		t.Fatalf("WriteTo error = %v, want the writer's error", err)
+	}
+	if n := len(a.requests()) + len(b.requests()); n != 1 {
+		t.Fatalf("replica requests = %d, want 1", n)
+	}
+	if _, err := vc.Get(vid); err != nil {
+		t.Fatalf("a client write error dropped the cached locations: %v", err)
+	}
+}
@@ -51,31 +51,41 @@ var (
 // authorization the master stamped on the volume; pass it through to the
 // volume server on the read request.
 func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) {
+	fullUrls, jwt, err := lookupFileIdUrls(masterFn, grpcDialOption, fileId)
+	if err != nil {
+		return "", jwt, err
+	}
+	return fullUrls[0], jwt, nil
+}
+
+// lookupFileIdUrls returns the read URLs of every replica of fileId, local
+// replicas first, each group in random order.
+func lookupFileIdUrls(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrls []string, jwt string, err error) {
 	parts := strings.Split(fileId, ",")
 	if len(parts) != 2 {
-		return "", jwt, errors.New("Invalid fileId " + fileId)
+		return nil, jwt, errors.New("Invalid fileId " + fileId)
 	}
 	lookup, lookupError := LookupVolumeId(masterFn, grpcDialOption, parts[0])
 	if lookupError != nil {
-		return "", jwt, lookupError
+		return nil, jwt, lookupError
 	}
 	if len(lookup.Locations) == 0 {
-		return "", jwt, errors.New("File Not Found")
+		return nil, jwt, errors.New("File Not Found")
 	}

-	localUrls := make([]string, 0, len(lookup.Locations))
+	var localUrls, remoteUrls []string
 	for _, loc := range lookup.Locations {
-		if !loc.DataInRemote {
-			localUrls = append(localUrls, loc.Url)
-		}
-	}
-	if len(localUrls) == 0 {
-		for _, loc := range lookup.Locations {
-			localUrls = append(localUrls, loc.Url)
+		fullUrl := "http://" + loc.Url + "/" + fileId
+		if loc.DataInRemote {
+			remoteUrls = append(remoteUrls, fullUrl)
+		} else {
+			localUrls = append(localUrls, fullUrl)
 		}
 	}
+	rand.Shuffle(len(localUrls), func(i, j int) { localUrls[i], localUrls[j] = localUrls[j], localUrls[i] })
+	rand.Shuffle(len(remoteUrls), func(i, j int) { remoteUrls[i], remoteUrls[j] = remoteUrls[j], remoteUrls[i] })

-	return "http://" + localUrls[rand.IntN(len(localUrls))] + "/" + fileId, lookup.Jwt, nil
+	return append(localUrls, remoteUrls...), lookup.Jwt, nil
 }

 func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) {