s3 sink: report the source read error the SDK hides (#11277)

* s3 sink: report the source read error the SDK hides

filer.backup stops for good on an event whose chunks are gone from the
volume servers: the uploader reads the body, the read fails with the
volume's 404, and the AWS SDK returns "ContentLength=N with Body length 0"
without the cause. isIgnorable404 would skip such an event, but it never
sees the 404, so the event is retried forever and the checkpoint never
advances.

ChunkStreamReader keeps its first source failure and the s3 sink returns
it when the upload fails.

* s3 sink: trim verbose comments on source error propagation

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/filer/stream.go

@@ -369,6 +369,7 @@ type ChunkStreamReader struct {
 	bufferLock   sync.Mutex
 	chunk        string
 	lookupFileId wdclient.LookupFileIdFunctionType
+	sourceErr    error
 }
 
 var _ = io.ReadSeeker(&ChunkStreamReader{})
@@ -507,7 +508,7 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 	urlStrings, err := c.lookupFileId(context.Background(), chunkView.FileId)
 	if err != nil {
 		glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
-		return err
+		return c.rememberSourceError(err)
 	}
 	var buffer bytes.Buffer
 	// pre-size to the known chunk size; avoids bytes.Buffer's doubling regrowth
@@ -529,7 +530,7 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 		}
 	}
 	if err != nil {
-		return err
+		return c.rememberSourceError(err)
 	}
 	c.buffer = buffer.Bytes()
 	c.bufferOffset = chunkView.ViewOffset
@@ -540,6 +541,28 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 	return nil
 }
 
+func (c *ChunkStreamReader) rememberSourceError(err error) error {
+	if c.sourceErr == nil {
+		c.sourceErr = err
+	}
+	return err
+}
+
+// SourceError returns the first lookup or chunk read failure, or nil.
+func (c *ChunkStreamReader) SourceError() error {
+	c.bufferLock.Lock()
+	defer c.bufferLock.Unlock()
+	return c.sourceErr
+}
+
+// ReaderSourceError returns the SourceError of a reader from NewFileReader, or nil.
+func ReaderSourceError(r io.Reader) error {
+	if csr, ok := r.(*ChunkStreamReader); ok {
+		return csr.SourceError()
+	}
+	return nil
+}
+
 func (c *ChunkStreamReader) Close() error {
 	c.bufferLock.Lock()
 	defer c.bufferLock.Unlock()

weed/filer/stream_source_error_test.go

@@ -0,0 +1,55 @@
+package filer
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+func TestChunkStreamReaderSourceError(t *testing.T) {
+	server := createTestServer(map[string][]byte{})
+	defer server.Close()
+
+	lookup := func(ctx context.Context, fileId string) ([]string, error) {
+		return []string{server.URL + "/" + fileId}, nil
+	}
+	reader := NewChunkStreamReaderFromLookup(context.Background(), lookup,
+		[]*filer_pb.FileChunk{{FileId: "7,01637037d6", Size: 8}})
+
+	if _, err := io.ReadAll(reader); err == nil {
+		t.Fatal("reading a gone chunk succeeded")
+	}
+	sourceErr := reader.SourceError()
+	if sourceErr == nil {
+		t.Fatal("reader kept no source error")
+	}
+	if !strings.Contains(sourceErr.Error(), "404") {
+		t.Fatalf("source error does not name the volume answer: %v", sourceErr)
+	}
+	if ReaderSourceError(reader) != sourceErr {
+		t.Fatal("ReaderSourceError does not return the kept error")
+	}
+}
+
+func TestChunkStreamReaderSourceErrorOnLookup(t *testing.T) {
+	master := &testMasterClient{urls: map[string][]string{}}
+	reader := NewChunkStreamReaderFromLookup(context.Background(), master.GetLookupFileIdFunction(),
+		[]*filer_pb.FileChunk{{FileId: "7,01637037d6", Size: 8}})
+
+	if _, err := io.ReadAll(reader); err == nil {
+		t.Fatal("reading a chunk with no location succeeded")
+	}
+	if reader.SourceError() == nil {
+		t.Fatal("reader kept no source error")
+	}
+}
+
+func TestReaderSourceErrorInlineEntry(t *testing.T) {
+	reader := NewFileReader(nil, &filer_pb.Entry{Content: []byte("inline")})
+	if err := ReaderSourceError(reader); err != nil {
+		t.Fatalf("inline content reported a source error: %v", err)
+	}
+}

weed/replication/sink/s3sink/s3_sink.go

@@ -220,6 +220,11 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
 		uploadInput.ContentMD5 = aws.String(base64.StdEncoding.EncodeToString([]byte(entry.Attributes.Md5)))
 	}
 	_, err = uploader.Upload(&uploadInput)
+	if err != nil {
+		if sourceErr := filer.ReaderSourceError(reader); sourceErr != nil {
+			return fmt.Errorf("read source %s: %w", key, sourceErr)
+		}
+	}
 
 	return err