From 42b0ca78506146edee2f964ea19f8ed83049a7dd Mon Sep 17 00:00:00 2001
From: Konstantin Adzer <47555435+AdzerKI@users.noreply.github.com>
Date: Sat, 12 Sep 2026 05:44:30 +0300
Subject: [PATCH] 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                    | 27 +++++++++++-
 weed/filer/stream_source_error_test.go  | 55 +++++++++++++++++++++++++
 weed/replication/sink/s3sink/s3_sink.go |  5 +++
 3 files changed, 85 insertions(+), 2 deletions(-)
 create mode 100644 weed/filer/stream_source_error_test.go

diff --git a/weed/filer/stream.go b/weed/filer/stream.go
index a910061f6..8642339c4 100644
--- a/weed/filer/stream.go
+++ b/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()
diff --git a/weed/filer/stream_source_error_test.go b/weed/filer/stream_source_error_test.go
new file mode 100644
index 000000000..cf6cb98be
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go
index 380a69d56..8f365a6e5 100644
--- a/weed/replication/sink/s3sink/s3_sink.go
+++ b/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