From 6eb2fb4842856c79823b68f998ac2d70ac1dda42 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 13 Nov 2026 20:17:46 -0800 Subject: [PATCH] filer: stop full-chunk reads at the chunk size after a mount truncate (#12632) * util/http: stop a full-chunk read at the chunk's recorded size weed mount shrinks a chunk's Size in place on ftruncate but leaves its needle untouched. A view covering the whole shortened chunk is fetched without a Range header, so the reader returned every stored byte. The filer GET then wrote past the declared Content-Length and sent an empty body, and the stream reader served the stale tail. ReadUrlAsStream and the cipher path now deliver at most size bytes; size 0 still reads the whole needle for manifest and log chunks. Co-Authored-By: Claude Opus 5.5 * s3api: limit a plain chunk read to the requested size A full-chunk read sends no Range header, so a chunk shortened by a mount truncate streamed its whole needle into the S3 response. Co-Authored-By: Claude Opus 5.5 * util/http: still report a failure past the chunk size The volume server aborts a full-needle response when the needle fails its checksum, and only at the end. Stopping at the chunk size missed that, so a damaged copy read as healthy. Read the rest of the body and return its error so the caller tries another copy. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3api_object_handlers.go | 2 +- ...3api_object_handlers_put_ambiguous_test.go | 30 +++++++ weed/util/http/http_global_client_util.go | 13 ++- .../util/http/http_global_client_util_test.go | 83 +++++++++++++++++++ 4 files changed, 125 insertions(+), 3 deletions(-) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 5580c344b1..ff62601268 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2325,7 +2325,7 @@ func (s3a *S3ApiServer) readChunkRange(ctx context.Context, fileId string, offse return nil, err } - return resp.Body, nil + return &rc{Reader: io.LimitReader(resp.Body, size), Closer: resp.Body}, nil } // getEncryptedStreamFromVolumes gets a raw encrypted data stream from volume diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 43d7e311b2..8f25ef38f6 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -1072,6 +1072,36 @@ func TestFetchChunkDataDecryptsVolumeCipher(t *testing.T) { } } +// A truncated chunk keeps its longer needle; full-chunk reads stop at the +// chunk's recorded size. +func TestFetchChunkDataStopsAtChunkSize(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{volume: volume, entries: map[string]*filer_pb.Entry{}} + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + + stored := bytes.Repeat([]byte("t"), 300) + fid := "3,01637037d6" + volume.mu.Lock() + volume.stored[fid] = stored + volume.mu.Unlock() + + view := &filer.ChunkView{FileId: fid, ViewSize: 100, ChunkSize: 100} + for name, fetch := range map[string]func(context.Context, *filer.ChunkView) (io.ReadCloser, error){ + "full": s3a.fetchFullChunk, + "view": s3a.fetchChunkViewData, + } { + r, err := fetch(context.Background(), view) + if err != nil { + t.Fatal(err) + } + got, _ := io.ReadAll(r) + r.Close() + if !bytes.Equal(got, stored[:100]) { + t.Fatalf("%s chunk read returned %d bytes, want 100", name, len(got)) + } + } +} + // A suspended-versioning overwrite resolves the conditional check against the // entry the read path serves: the null slot when present, else the latest // version under .versions/. The routed transaction carries that version as the diff --git a/weed/util/http/http_global_client_util.go b/weed/util/http/http_global_client_util.go index cf492230b0..c21e3df59e 100644 --- a/weed/util/http/http_global_client_util.go +++ b/weed/util/http/http_global_client_util.go @@ -430,6 +430,11 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte, default: reader = r.Body } + var body io.Reader = reader + if size > 0 { + // a truncated chunk's needle can hold more than the chunk's recorded size + body = io.LimitReader(reader, int64(size)) + } var ( m int @@ -445,11 +450,15 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte, default: } - m, err = reader.Read(buf) + m, err = body.Read(buf) if m > 0 { fn(buf[:m]) } if err == io.EOF { + // a needle that fails its checksum aborts the response at the end + if _, err = io.Copy(io.Discard, reader); err != nil { + return true, err + } return false, nil } if err != nil { @@ -477,7 +486,7 @@ func readEncryptedUrl(ctx context.Context, fileUrl, jwt string, cipherKey []byte if len(decryptedData) < int(offset)+size { return false, fmt.Errorf("read decrypted %s size %d [%d, %d)", fileUrl, len(decryptedData), offset, int(offset)+size) } - if isFullChunk { + if isFullChunk && size == 0 { fn(decryptedData) } else { sliceEnd := int(offset) + size diff --git a/weed/util/http/http_global_client_util_test.go b/weed/util/http/http_global_client_util_test.go index c5e5f32cc0..318387efe9 100644 --- a/weed/util/http/http_global_client_util_test.go +++ b/weed/util/http/http_global_client_util_test.go @@ -8,6 +8,8 @@ import ( "net/http/httptest" "testing" "time" + + "github.com/seaweedfs/seaweedfs/weed/util" ) func TestAppendQueryParameter(t *testing.T) { @@ -94,6 +96,87 @@ func TestReadUrlAsStreamReturnsGzipReaderError(t *testing.T) { } } +// A truncated chunk keeps its needle, so a full-chunk read must stop at the +// chunk's recorded size instead of returning every stored byte. +func TestReadUrlAsStreamStopsAtChunkSize(t *testing.T) { + InitGlobalHttpClient() + + stored := bytes.Repeat([]byte("t"), 300) + cipherKey := util.GenCipherKey() + encrypted, err := util.Encrypt(stored, cipherKey) + if err != nil { + t.Fatal(err) + } + var gzipped bytes.Buffer + zw := gzip.NewWriter(&gzipped) + zw.Write(stored) + zw.Close() + + for _, tc := range []struct { + name string + body []byte + encoding string + cipherKey []byte + }{ + {name: "plain", body: stored}, + {name: "gzip", body: gzipped.Bytes(), encoding: "gzip"}, + {name: "cipher", body: encrypted, cipherKey: cipherKey}, + } { + t.Run(tc.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if tc.encoding != "" { + w.Header().Set("Content-Encoding", tc.encoding) + } + _, _ = w.Write(tc.body) + })) + defer server.Close() + + var got []byte + if _, err := ReadUrlAsStream(context.Background(), server.URL, "", tc.cipherKey, false, true, 0, 100, func(data []byte) { + got = append(got, data...) + }); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, stored[:100]) { + t.Fatalf("full-chunk read returned %d bytes, want 100", len(got)) + } + + got = nil + if _, err := ReadUrlAsStream(context.Background(), server.URL, "", tc.cipherKey, false, true, 0, 0, func(data []byte) { + got = append(got, data...) + }); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, stored) { + t.Fatalf("whole-needle read returned %d bytes, want %d", len(got), len(stored)) + } + }) + } +} + +func TestReadUrlAsStreamReportsAbortPastChunkSize(t *testing.T) { + InitGlobalHttpClient() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "300") + _, _ = w.Write(bytes.Repeat([]byte("t"), 200)) + w.(http.Flusher).Flush() + panic(http.ErrAbortHandler) + })) + defer server.Close() + + var got int + retryable, err := ReadUrlAsStream(context.Background(), server.URL, "", nil, false, true, 0, 100, func(data []byte) { + got += len(data) + }) + if err == nil || !retryable { + t.Fatalf("aborted response read as retryable=%v err=%v, want a retryable error", retryable, err) + } + if got != 100 { + t.Fatalf("delivered %d bytes, want 100", got) + } +} + func TestReadErrorsCarryTheServerMessage(t *testing.T) { InitGlobalHttpClient()