From 8efa32258afa68688ace8e8d040c2501ae996b88 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 10 May 2026 15:57:07 -0700 Subject: [PATCH] feat(volume): X-Seaweedfs-Original-Size hint skips redundant gunzip (#9426) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(volume): X-Seaweedfs-Original-Size hint skips redundant gunzip The full-chunk gzip pass-through (#9425) fixed source-volume decompression but moved the cost to the destination volume: parseUpload still ran util.DecompressData on the forwarded gzipped bytes, just to learn the uncompressed length so it could record OriginalDataSize in the needle metadata. For 6-way concurrent 64 MiB UploadPartCopy that decompress-and-discard pass dominated the remaining heap profile after the streaming chain landed (~297 MiB inuse via bytes.Buffer.ReadFrom inside util.GunzipStream). Add an X-Seaweedfs-Original-Size header on the multipart part. When the upstream sets it (the s3 chunk-copy fast path always knows the uncompressed size — it's the source chunk's logical size) and no Content-MD5 verification is requested (which would require decompressed bytes to compute against), parseUpload uses the hint directly and skips the decompress. Header is X-* prefixed (not Seaweed-*) so it doesn't get auto-stored as a needle pair by PairNamePrefix. Backward compatible: - old s3 servers don't set the header, parseUpload decompresses as before - new s3 servers talking to old volumes: header is ignored, volume decompresses - bad header values (non-numeric, negative, garbage) fall back to the existing decompress path End-to-end repro impact (512 MiB src, 6 parallel UploadPartCopy, post-#9420/#9421/#9422/#9424/#9425 baseline): RSS, round 2: 1149 MiB → 594 MiB heap inuse_space: 545 MiB → 349 MiB HeapSys: 1.35 GiB → 777 MiB TotalAlloc cum: ~9 GiB → 3.5 GiB Total reduction from pre-#9420 baseline: 3134 → 594 MiB (-81%). Test exercises the four matrix corners (hint+no-MD5, hint+part-MD5, hint+req-MD5, no-hint, garbage-hint) and bounds allocation per case so a regression that re-introduces the unconditional decompress fails the hint-present-no-MD5 case. * review: bound X-Seaweedfs-Original-Size by sizeLimit and uint32 CodeQL on PR 9426 traced a new taint flow: the strconv.Atoi(hint) value flows into pu.OriginalDataSize -> originalSize -> the existing uint32(originalSize) cast in volume_server_handlers_write.go:73. The cast was always there but its input was previously bounded by the ParseUpload read path (capped at sizeLimit). Adding a user-controlled hint bypassed that bound, so a malicious header could overflow the uint32 silently. Bound the hint at parse time by sizeLimit (the largest needle this volume will accept anyway) and by math.MaxUint32 (belt-and-suspenders in case sizeLimit is configured > 4 GiB). --- weed/storage/needle/needle.go | 9 + weed/storage/needle/needle_parse_upload.go | 35 +++- .../needle_parse_upload_size_hint_test.go | 186 ++++++++++++++++++ 3 files changed, 229 insertions(+), 1 deletion(-) create mode 100644 weed/storage/needle/needle_parse_upload_size_hint_test.go diff --git a/weed/storage/needle/needle.go b/weed/storage/needle/needle.go index 7c0ce1c20..bf8231401 100644 --- a/weed/storage/needle/needle.go +++ b/weed/storage/needle/needle.go @@ -16,6 +16,15 @@ import ( const ( NeedleChecksumSize = 4 PairNamePrefix = "Seaweed-" + + // OriginalSizeHeader is set on multipart parts whose body is already + // gzip-encoded by an upstream component (e.g. the s3 chunk-copy fast + // path forwarding compressed bytes). It declares the uncompressed + // length so parseUpload can record OriginalDataSize without paying a + // decompress-then-discard pass just to learn the size — see #6541. + // The prefix is X-* (not Seaweed-*) so PairNamePrefix's auto-record + // of Seaweed-prefixed headers as needle pairs doesn't pick this up. + OriginalSizeHeader = "X-Seaweedfs-Original-Size" ) /* diff --git a/weed/storage/needle/needle_parse_upload.go b/weed/storage/needle/needle_parse_upload.go index aa54c3462..a4f2d9559 100644 --- a/weed/storage/needle/needle_parse_upload.go +++ b/weed/storage/needle/needle_parse_upload.go @@ -6,6 +6,7 @@ import ( "encoding/base64" "fmt" "io" + "math" "mime" "net/http" "path" @@ -56,6 +57,11 @@ type ParsedUpload struct { IsChunkedFile bool UncompressedData []byte ContentMd5 string + // originalSizeHint is the value parsed from OriginalSizeHeader on the + // multipart part. If positive and IsGzipped is true, ParseUpload skips + // the size-learning DecompressData pass — we already know the + // uncompressed length. + originalSizeHint int } func ParseUpload(r *http.Request, sizeLimit int64, bytesBuffer *bytes.Buffer) (pu *ParsedUpload, e error) { @@ -83,7 +89,23 @@ func ParseUpload(r *http.Request, sizeLimit int64, bytesBuffer *bytes.Buffer) (p pu.UncompressedData = pu.Data // println("received data", len(pu.Data), "isGzipped", pu.IsGzipped, "mime", pu.MimeType, "name", pu.FileName) if pu.IsGzipped { - if unzipped, e := util.DecompressData(pu.Data); e == nil { + // If the upstream sent X-Seaweedfs-Original-Size and we don't need + // the uncompressed bytes for a Content-MD5 check, skip the + // decompress-and-discard pass that would otherwise just re-derive + // the size we were already told. For a 64 MiB chunk that pass + // allocates ~128 MiB through bytes.Buffer.ReadFrom's geometric + // grow inside util.GunzipStream — see #6541. + expectedChecksum := r.Header.Get("Content-MD5") + if expectedChecksum == "" { + expectedChecksum = pu.ContentMd5 + } + if pu.originalSizeHint > 0 && expectedChecksum == "" { + pu.OriginalDataSize = pu.originalSizeHint + // pu.UncompressedData stays as pu.Data (the gzipped bytes). + // Downstream consumers either don't read it (chunk-copy + // uploads don't) or fall back to decompress-on-demand if they + // need the raw bytes. + } else if unzipped, e := util.DecompressData(pu.Data); e == nil { pu.OriginalDataSize = len(unzipped) pu.UncompressedData = unzipped // println("ungzipped data size", len(unzipped)) @@ -211,6 +233,17 @@ func parseUpload(r *http.Request, sizeLimit int64, pu *ParsedUpload) (e error) { pu.IsGzipped = part.Header.Get("Content-Encoding") == "gzip" // pu.IsZstd = part.Header.Get("Content-Encoding") == "zstd" + if hint := part.Header.Get(OriginalSizeHeader); hint != "" { + // Bound the user-controlled value: it flows into needle + // metadata (OriginalDataSize, then uint32(originalSize) at the + // volume-server caller). Cap at sizeLimit (the largest needle + // this volume will accept anyway) and at math.MaxUint32 to + // keep the downstream cast safe even if a future deployment + // configures a multi-GiB sizeLimit. + if n, err := strconv.Atoi(hint); err == nil && n > 0 && int64(n) <= sizeLimit && n <= math.MaxUint32 { + pu.originalSizeHint = n + } + } } else { disposition := r.Header.Get("Content-Disposition") diff --git a/weed/storage/needle/needle_parse_upload_size_hint_test.go b/weed/storage/needle/needle_parse_upload_size_hint_test.go new file mode 100644 index 000000000..a367f94bd --- /dev/null +++ b/weed/storage/needle/needle_parse_upload_size_hint_test.go @@ -0,0 +1,186 @@ +package needle + +import ( + "bytes" + "compress/gzip" + "crypto/md5" + "encoding/base64" + "mime/multipart" + "net/http" + "net/http/httptest" + "net/textproto" + "runtime" + "strconv" + "testing" +) + +// TestParseUpload_OriginalSizeHint covers the gzipped-pass-through path +// added for https://github.com/seaweedfs/seaweedfs/issues/6541. When the +// upstream sets X-Seaweedfs-Original-Size on the multipart part, ParseUpload +// must skip the size-learning DecompressData pass, but only when no +// Content-MD5 verification is needed (MD5 must be computed against the +// uncompressed bytes to match what the s3 client sent). +func TestParseUpload_OriginalSizeHint(t *testing.T) { + const uncompressedSize = 4 * 1024 * 1024 // 4 MiB + uncompressed := make([]byte, uncompressedSize) + for i := range uncompressed { + uncompressed[i] = byte(i*31 + 7) + } + + gzipped := gzipBytes(t, uncompressed) + uncompressedMD5 := base64.StdEncoding.EncodeToString(md5sum(uncompressed)) + + cases := []struct { + name string + hintBytes int64 // 0 means don't send the header + md5OnPart string + md5OnReq string + wantSkip bool // ParseUpload should NOT decompress + wantOrigSz int // expected pu.OriginalDataSize + wantErr string // substring; "" means no error + }{ + { + name: "hint present, no MD5 → skip decompress", + hintBytes: uncompressedSize, + wantSkip: true, + wantOrigSz: uncompressedSize, + }, + { + name: "no hint → existing decompress path", + wantSkip: false, + wantOrigSz: uncompressedSize, + }, + { + name: "hint present, MD5 on part → must decompress for MD5", + hintBytes: uncompressedSize, + md5OnPart: uncompressedMD5, + wantSkip: false, + wantOrigSz: uncompressedSize, + }, + { + name: "hint present, MD5 on request → must decompress for MD5", + hintBytes: uncompressedSize, + md5OnReq: uncompressedMD5, + wantSkip: false, + wantOrigSz: uncompressedSize, + }, + { + name: "garbage hint value falls back to decompress", + hintBytes: -1, // sentinel: write a non-numeric value below + wantSkip: false, + wantOrigSz: uncompressedSize, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + req := buildReq(t, gzipped, c.hintBytes, c.md5OnPart, c.md5OnReq) + + runtime.GC() + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + + bb := &bytes.Buffer{} + pu, err := ParseUpload(req, 256<<20, bb) + + runtime.ReadMemStats(&after) + allocated := after.TotalAlloc - before.TotalAlloc + + if c.wantErr != "" { + if err == nil { + t.Fatalf("ParseUpload: want error containing %q, got nil", c.wantErr) + } + return + } + if err != nil { + t.Fatalf("ParseUpload: %v", err) + } + if pu.OriginalDataSize != c.wantOrigSz { + t.Errorf("OriginalDataSize=%d, want %d", pu.OriginalDataSize, c.wantOrigSz) + } + + // Bound check: if we expected to skip decompress, the test should + // allocate well under uncompressedSize. If we expected to + // decompress, the test should allocate at least uncompressedSize. + if c.wantSkip { + bound := uint64(uncompressedSize) // generous + if allocated > bound { + t.Errorf("hint-present case allocated %d bytes for a %d-byte gzipped body; "+ + "bound %d (regression: did the decompress-skip break?)", + allocated, len(gzipped), bound) + } + } else { + // Decompress path must allocate at least the uncompressed + // size (the unzipped slice). Sanity-check it actually ran. + min := uint64(uncompressedSize) + if allocated < min { + t.Errorf("decompress-path case allocated only %d bytes; "+ + "expected at least %d (was the decompress wrongly skipped?)", + allocated, min) + } + } + t.Logf("allocated=%d bytes (gzipped=%d, uncompressed=%d)", allocated, len(gzipped), uncompressedSize) + }) + } +} + +// gzipBytes returns the gzip encoding of in. +func gzipBytes(t *testing.T, in []byte) []byte { + t.Helper() + var buf bytes.Buffer + gw := gzip.NewWriter(&buf) + if _, err := gw.Write(in); err != nil { + t.Fatalf("gzip write: %v", err) + } + if err := gw.Close(); err != nil { + t.Fatalf("gzip close: %v", err) + } + return buf.Bytes() +} + +func md5sum(in []byte) []byte { + h := md5.New() + h.Write(in) + return h.Sum(nil) +} + +// buildReq constructs an *http.Request whose body is a multipart/form-data +// body containing one file part. hintBytes>0 sets X-Seaweedfs-Original-Size +// to that value; hintBytes==-1 sets it to a non-numeric string (garbage +// case); hintBytes==0 omits the header. +func buildReq(t *testing.T, body []byte, hintBytes int64, md5OnPart, md5OnReq string) *http.Request { + t.Helper() + + var bodyBuf bytes.Buffer + mw := multipart.NewWriter(&bodyBuf) + h := make(textproto.MIMEHeader) + h.Set("Content-Disposition", `form-data; name="file"; filename="test.bin"`) + h.Set("Content-Encoding", "gzip") + switch { + case hintBytes > 0: + h.Set(OriginalSizeHeader, strconv.FormatInt(hintBytes, 10)) + case hintBytes == -1: + h.Set(OriginalSizeHeader, "not-a-number") + } + if md5OnPart != "" { + h.Set("Content-MD5", md5OnPart) + } + fw, err := mw.CreatePart(h) + if err != nil { + t.Fatalf("CreatePart: %v", err) + } + if _, err := fw.Write(body); err != nil { + t.Fatalf("write part: %v", err) + } + if err := mw.Close(); err != nil { + t.Fatalf("close mw: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/upload", bytes.NewReader(bodyBuf.Bytes())) + req.Header.Set("Content-Type", mw.FormDataContentType()) + req.ContentLength = int64(bodyBuf.Len()) + if md5OnReq != "" { + req.Header.Set("Content-MD5", md5OnReq) + } + return req +}