feat(mount): attach Content-MD5 to chunk uploads (#10016)

* mount: attach Content-MD5 to chunk uploads

Mount writes never set UploadOption.Md5, so FileChunk.ETag stays empty
and filer.ETag() degenerates to md5("")-N: same-size files compare
equal regardless of content, defeating metadata-level verification
like filer.sync.verify.

Compute each chunk's MD5 and send it as Content-MD5. The volume server
verifies it on ingest (rejecting in-flight corruption) and echoes it
back, persisting FileChunk.ETag like filer/S3 writes already do.

The dirty-page flush paths already pass a *util.BytesReader whose
backing slice is the whole chunk, so the digest is taken in place with
no extra read, copy, or allocation (UploadWithRetry unwraps it the same
way downstream). Only the rarer plain-reader callers (e.g. manifest
chunks) fall back to io.ReadAll. Skipped under -cipher, where only the
ciphertext reaches the server.

The digest encoding (std-base64 of the raw md5) is the contract the
volume server verifies against, so it is factored into contentMD5Base64
and covered by a unit test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* mount: compute chunk Content-MD5 in the uploader, not the caller

Move the WantMd5 hashing into UploadWithRetry, where the chunk is already
buffered for the retry path, so saveDataAsChunk stops type-switching the
reader and re-reading plain readers. One materialization point, and the
cipher exclusion lives next to the hash instead of at every call site.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/mount/weedfs_write.go

@@ -26,12 +26,14 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun
 			return
 		}
 
+		// WantMd5 gives mount writes a real filer.ETag (server echoes Content-MD5 back).
 		uploadOption := &operation.UploadOption{
 			Filename:          filename,
 			Cipher:            wfs.option.Cipher,
 			IsInputCompressed: false,
 			MimeType:          "",
 			PairMap:           nil,
+			WantMd5:           true,
 		}
 		genFileUrlFn := func(host, fileId string) string {
 			fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)

weed/operation/upload_content.go

@@ -3,6 +3,8 @@ package operation
 import (
 	"bytes"
 	"context"
+	"crypto/md5"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -39,6 +41,7 @@ type UploadOption struct {
 	Jwt               security.EncodedJwt
 	RetryForever      bool
 	Md5               string
+	WantMd5           bool // compute Content-MD5 from the data when Md5 is unset and the upload is not ciphered
 	BytesBuffer       *bytes.Buffer
 	SourceUrl         string // optional: for logging when reading from a remote source
 	MaxAttempts       int    // <=0 uses the default
@@ -196,6 +199,14 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi
 		assignRequest.ExpectedDataSize = uint64(len(data))
 	}
 
+	// Hash the buffer we already hold so the server echoes Content-MD5 back as
+	// the chunk ETag (std-base64 of the raw digest, the form ParseUpload
+	// verifies). Never under cipher: the server sees only ciphertext.
+	if uploadOption.WantMd5 && uploadOption.Md5 == "" && !uploadOption.Cipher {
+		digest := md5.Sum(data)
+		uploadOption.Md5 = base64.StdEncoding.EncodeToString(digest[:])
+	}
+
 	fileId, uploadResult, err = uploader.uploadWithRetryData(func() (fileId string, host string, auth security.EncodedJwt, err error) {
 		// grpc assign volume
 		if grpcAssignErr := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {