From 94357ac6a962cbdc27ce8c4b11d4b6cc3af0b394 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 13 Jun 2026 21:52:59 -0700 Subject: [PATCH] [volume] preserve compression state during replication (#9946) * preserve compression state during replication * explain why ParseUpload skips compression for replica writes * fix data race on err result in FetchAndWriteNeedle The local-write and replica-write goroutines all wrote the named err return under an unsynchronized err==nil check. Give each goroutine its own error slot and combine after wg.Wait(): local error wins, then the first replica failure. * skip redundant decompression of compressed needles during replication doUploadData decompressed a compressed input only to report the clear-data length on UploadResult.Size, which both replication callers discard. Skip the decompress when IsReplication. --- weed/operation/upload_content.go | 7 +-- weed/operation/upload_content_test.go | 56 ++++++++++++++++++++++ weed/server/volume_grpc_remote.go | 29 +++++++---- weed/storage/needle/needle_parse_upload.go | 3 +- weed/topology/store_replicate.go | 1 + 5 files changed, 82 insertions(+), 14 deletions(-) diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 6f98846fa5..1920478b49 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -33,6 +33,7 @@ type UploadOption struct { Filename string Cipher bool IsInputCompressed bool + IsReplication bool // preserve the source needle's compression state MimeType string PairMap map[string]string Jwt security.EncodedJwt @@ -276,7 +277,7 @@ func (uploader *Uploader) retriedUploadData(ctx context.Context, data []byte, op func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option *UploadOption) (uploadResult *UploadResult, err error) { contentIsGzipped := option.IsInputCompressed shouldGzipNow := false - if !option.IsInputCompressed { + if !option.IsInputCompressed && !option.IsReplication { if option.MimeType == "" { option.MimeType = http.DetectContentType(data) // println("detect1 mimetype to", MimeType) @@ -311,8 +312,8 @@ func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option contentIsGzipped = true } } - } else if option.IsInputCompressed { - // just to get the clear data length + } else if option.IsInputCompressed && !option.IsReplication { + // decompress only to report the clear data length; replication discards the result, so skip it clearData, err = util.DecompressData(data) if err == nil { clearDataLen = len(clearData) diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index 7694eecdcf..b9d925e899 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -7,12 +7,14 @@ import ( "fmt" "io" "net/http" + "net/http/httptest" "strings" "sync" "testing" "time" "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) type scriptedHTTPResponse struct { @@ -200,6 +202,60 @@ func TestUploadRewindsBodyOnConnectionReset(t *testing.T) { } } +func TestReplicationUploadPreservesUncompressedNeedle(t *testing.T) { + testCases := []struct { + name string + payload []byte + }{ + { + name: "unknown binary sampling", + payload: bytes.Repeat([]byte{0}, 32*1024), + }, + { + name: "receiver MIME detection", + payload: bytes.Repeat([]byte("plain text content\n"), 2048), + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + var replicatedNeedle *needle.Needle + var parseErr error + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + replicatedNeedle, _, _, parseErr = needle.CreateNeedleFromRequest(r, false, 1024*1024, &bytes.Buffer{}) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _, _ = io.WriteString(w, `{"name":"payload.custom","size":32768}`) + })) + defer server.Close() + + uploader := newUploader(server.Client()) + _, err := uploader.UploadData(context.Background(), tc.payload, &UploadOption{ + UploadUrl: server.URL + "/3,01637037d6?type=replicate", + Filename: "payload.custom", + IsReplication: true, + MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("replication upload failed: %v", err) + } + if parseErr != nil { + t.Fatalf("parse replicated upload: %v", parseErr) + } + if replicatedNeedle == nil { + t.Fatal("replica did not receive a needle") + } + if replicatedNeedle.IsCompressed() { + t.Fatal("replication changed an uncompressed needle to compressed") + } + if !bytes.Equal(replicatedNeedle.Data, tc.payload) { + t.Fatalf("replicated data differs: got %d bytes, want %d", len(replicatedNeedle.Data), len(tc.payload)) + } + }) + } +} + // deadReplicaClient simulates an unreachable replica: every request blocks for // dialDelay (standing in for the TCP dial timeout) and then fails, unless the // request context is cancelled first. diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index 1be5fc3eb5..6519523006 100644 --- a/weed/server/volume_grpc_remote.go +++ b/weed/server/volume_grpc_remote.go @@ -237,6 +237,8 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser } var wg sync.WaitGroup + var localErr error + replicaErrs := make([]error, len(req.Replicas)) wg.Add(1) go func() { defer wg.Done() @@ -250,43 +252,50 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser n.LastModified = uint64(time.Now().Unix()) n.SetHasLastModifiedDate() if _, localWriteErr := vs.store.WriteVolumeNeedle(v.Id, n, true, false); localWriteErr != nil { - if err == nil { - err = fmt.Errorf("local write needle %d size %d: %v", req.NeedleId, req.Size, localWriteErr) - } + localErr = fmt.Errorf("local write needle %d size %d: %v", req.NeedleId, req.Size, localWriteErr) } else { resp.ETag = n.Etag() } }() if len(req.Replicas) > 0 { fileId := needle.NewFileId(v.Id, req.NeedleId, req.Cookie) - for _, replica := range req.Replicas { + for i, replica := range req.Replicas { wg.Add(1) - go func(targetVolumeServer string) { + go func(idx int, targetVolumeServer string) { defer wg.Done() uploadOption := &operation.UploadOption{ UploadUrl: fmt.Sprintf("http://%s/%s?type=replicate", targetVolumeServer, fileId.String()), Filename: "", Cipher: false, IsInputCompressed: false, + IsReplication: true, MimeType: "", PairMap: nil, Jwt: security.EncodedJwt(req.Auth), } uploader, uploaderErr := operation.NewUploader() - if uploaderErr != nil && err == nil { - err = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, uploaderErr) + if uploaderErr != nil { + replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, uploaderErr) return } - if _, replicaWriteErr := uploader.UploadData(ctx, data, uploadOption); replicaWriteErr != nil && err == nil { - err = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, replicaWriteErr) + if _, replicaWriteErr := uploader.UploadData(ctx, data, uploadOption); replicaWriteErr != nil { + replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, replicaWriteErr) } - }(replica.Url) + }(i, replica.Url) } } wg.Wait() + // local write error wins; otherwise surface the first replica failure + err = localErr + for _, replicaErr := range replicaErrs { + if err == nil { + err = replicaErr + } + } + return resp, err } diff --git a/weed/storage/needle/needle_parse_upload.go b/weed/storage/needle/needle_parse_upload.go index fc4c5637e9..99b4e13ca4 100644 --- a/weed/storage/needle/needle_parse_upload.go +++ b/weed/storage/needle/needle_parse_upload.go @@ -94,7 +94,8 @@ func ParseUpload(r *http.Request, sizeLimit int64, bytesBuffer *bytes.Buffer) (p } else if n, err := util.GunzipStream(io.Discard, bytes.NewReader(pu.Data)); err == nil { pu.OriginalDataSize = int(n) } - } else { + } else if r.URL.Query().Get("type") != "replicate" { + // replica writes must keep the source needle's compression state, not re-derive it ext := filepath.Base(pu.FileName) mimeType := pu.MimeType if mimeType == "" { diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index 147a1383d4..18d329ba59 100644 --- a/weed/topology/store_replicate.go +++ b/weed/topology/store_replicate.go @@ -110,6 +110,7 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi Filename: string(n.Name), Cipher: false, IsInputCompressed: n.IsCompressed(), + IsReplication: true, MimeType: string(n.Mime), PairMap: pairMap, Jwt: jwt,