[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

@@ -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)

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.

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
 }

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 == "" {

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,