mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 00:09:00 +01:00
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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
63c99183d1
commit
6eb2fb4842
4 files changed
+125
-3
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
Reference in new issue
Block a user