refactor: centralize genUploadUrl in UploadOption (#10164)

* refactor: centralize genUploadUrl in UploadOption

Replace inline genFileUrlFn closures with operation.GenUploadUrl field:

- Add GenUploadUrl func(host, fileId) string to UploadOption struct
- Add GenUploadUrlProxy(filerAddress string) utility function
- Remove genFileUrlFn parameter from UploadWithRetry signature
- Update all callers: mount, gateway, mq, filer_copy, filer_sync

This matches the weed mount -filerProxy pattern exactly,
factorizing the URL generation logic across all consumers.

Co-Authored-By: Athena 🏛️ <hermes-agent@local> (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf)

* docker release: run all platform jobs in one wave, cache rocksdb compile

Drop max-parallel so the 13 per-platform builds run together instead of two
waves of 8 (rocksdb was queuing behind the cap and starting ~8 min late).

Keep cache-to mode=max for rocksdb: its RocksDB static_lib compile is
sha-independent, so it caches across releases and stops being the ~16-min
long-pole that gates the merge fan-in. go-build variants stay mode=min.

Co-Authored-By: Athena 🏛️ <hermes-agent@local> (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf)

* refactor: centralize genUploadUrl in UploadOption

Replace inline genFileUrlFn closures with operation.GenUploadUrl field:

- Add GenUploadUrl func(host, fileId) string to UploadOption struct
- Add GenUploadUrlProxy(filerAddress string) utility function
- Remove genFileUrlFn parameter from UploadWithRetry signature
- Update all callers: mount, gateway, mq, filer_copy, filer_sync

This matches the weed mount -filerProxy pattern exactly,
factorizing the URL generation logic across all consumers.

Co-Authored-By: Athena 🏛️ <hermes-agent@local> (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf)

* Remove accidental ROCmFPX submodule reference

* gofmt chunk upload option block

* Preserve broker cipher and re-read proxy filer per upload attempt

Chunk uploads must keep the configured Cipher, and both the mount and broker current filer can change on failover, so build the proxy upload URL inside the closure instead of capturing the address once.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/command/filer_copy.go

@@ -371,9 +371,6 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
 				MimeType:          mimeType,
 				PairMap:           nil,
 			},
-			func(host, fileId string) string {
-				return fmt.Sprintf("http://%s/%s", host, fileId)
-			},
 			util.NewBytesReader(data),
 		)
 		if flushErr != nil {
@@ -456,9 +453,6 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
 					MimeType:          "",
 					PairMap:           nil,
 				},
-				func(host, fileId string) string {
-					return fmt.Sprintf("http://%s/%s", host, fileId)
-				},
 				io.NewSectionReader(f, i*chunkSize, chunkSize),
 			)
 
@@ -576,9 +570,6 @@ func (worker *FileCopyWorker) saveDataAsChunk(reader io.Reader, name string, off
 			MimeType:          "",
 			PairMap:           nil,
 		},
-		func(host, fileId string) string {
-			return fmt.Sprintf("http://%s/%s", host, fileId)
-		},
 		reader,
 	)
 

weed/filer/gateway_upload.go

@@ -17,7 +17,6 @@ type GatewayChunkUploader interface {
 		filerClient filer_pb.FilerClient,
 		assignRequest *filer_pb.AssignVolumeRequest,
 		uploadOption *operation.UploadOption,
-		genFileUrlFn func(host, fileId string) string,
 		reader io.Reader,
 	) (fileId string, uploadResult *operation.UploadResult, err error, data []byte)
 }
@@ -120,11 +119,8 @@ func SaveGatewayDataAsChunk(req GatewayChunkUploadRequest) (*filer_pb.FileChunk,
 		PairMap:           req.PairMap,
 	}
 
-	genFileUrlFn := func(host, fileId string) string {
-		if req.VolumeServerAccess == "filerProxy" && req.FilerHTTPAddress != "" {
-			return fmt.Sprintf("http://%s/?proxyChunkId=%s", req.FilerHTTPAddress, fileId)
-		}
-		return fmt.Sprintf("http://%s/%s", host, fileId)
+	if req.VolumeServerAccess == "filerProxy" && req.FilerHTTPAddress != "" {
+		uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(req.FilerHTTPAddress)
 	}
 
 	assignRequest := &filer_pb.AssignVolumeRequest{
@@ -141,7 +137,6 @@ func SaveGatewayDataAsChunk(req GatewayChunkUploadRequest) (*filer_pb.FileChunk,
 		req.FilerClient,
 		assignRequest,
 		uploadOption,
-		genFileUrlFn,
 		req.Reader,
 	)
 	if uploadErr != nil {

weed/mount/weedfs_write.go

@@ -35,12 +35,11 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun
 			PairMap:           nil,
 			WantMd5:           true,
 		}
-		genFileUrlFn := func(host, fileId string) string {
-			fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
-			if wfs.option.VolumeServerAccess == "filerProxy" {
-				fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId)
+		if wfs.option.VolumeServerAccess == "filerProxy" {
+			// getCurrentFiler() can change on failover, so read it per attempt.
+			uploadOption.GenUploadUrl = func(host, fileId string) string {
+				return fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId)
 			}
-			return fileUrl
 		}
 
 		fileId, uploadResult, err, data := uploader.UploadWithRetry(
@@ -54,7 +53,7 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun
 				DataCenter:  wfs.option.DataCenter,
 				Path:        assignPath,
 			},
-			uploadOption, genFileUrlFn, reader,
+			uploadOption, reader,
 		)
 
 		if err != nil {

weed/mq/broker/broker_write.go

@@ -161,27 +161,26 @@ func (b *MessageQueueBroker) assignAndUpload(targetFile string, data []byte) (fi
 		return
 	}
 
+	uploadOption := &operation.UploadOption{
+		Cipher: b.option.Cipher,
+	}
+	if b.option.VolumeServerAccess == "filerProxy" {
+		// b.currentFiler can change on failover, so read it per attempt.
+		uploadOption.GenUploadUrl = func(host, fileId string) string {
+			return fmt.Sprintf("http://%s/?proxyChunkId=%s", b.currentFiler, fileId)
+		}
+	}
+
 	fileId, uploadResult, err, _ = uploader.UploadWithRetry(
 		b,
 		&filer_pb.AssignVolumeRequest{
 			Count:       1,
 			Replication: b.option.DefaultReplication,
 			Collection:  "topics",
-			// TtlSec:      wfs.option.TtlSec,
-			// DiskType:    string(wfs.option.DiskType),
-			DataCenter: b.option.DataCenter,
-			Path:       targetFile,
-		},
-		&operation.UploadOption{
-			Cipher: b.option.Cipher,
-		},
-		func(host, fileId string) string {
-			fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
-			if b.option.VolumeServerAccess == "filerProxy" {
-				fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", b.currentFiler, fileId)
-			}
-			return fileUrl
+			DataCenter:  b.option.DataCenter,
+			Path:        targetFile,
 		},
+		uploadOption,
 		reader,
 	)
 	return

weed/mq/logstore/log_to_parquet.go

@@ -479,9 +479,6 @@ func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile
 				MimeType:          "application/vnd.apache.parquet",
 				PairMap:           nil,
 			},
-			func(host, fileId string) string {
-				return fmt.Sprintf("http://%s/%s", host, fileId)
-			},
 			io.NewSectionReader(sourceFile, i*chunkSize, chunkSize),
 		)
 		if err != nil {

weed/operation/upload_content.go

@@ -30,6 +30,21 @@ import (
 	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 )
 
+// GenUploadUrlProxy returns a function that builds a chunk upload URL via the
+// filer proxy.  Identical to the inline logic used by weed mount -filerProxy
+// and weed filer gateway:
+//
+//   - without filerProxy: "http://{host}/{fileId}"
+//   - with filerProxy:     "http://{filerAddress}/?proxyChunkId={fileId}"
+func GenUploadUrlProxy(filerAddress string) func(host, fileId string) string {
+	return func(host, fileId string) string {
+		if filerAddress == "" {
+			return fmt.Sprintf("http://%s/%s", host, fileId)
+		}
+		return fmt.Sprintf("http://%s/?proxyChunkId=%s", filerAddress, fileId)
+	}
+}
+
 type UploadOption struct {
 	UploadUrl         string
 	Filename          string
@@ -43,8 +58,9 @@ type UploadOption struct {
 	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
+	SourceUrl         string                           // optional: for logging when reading from a remote source
+	MaxAttempts       int                              // <=0 uses the default
+	GenUploadUrl      func(host, fileId string) string // if nil → fallback "http://{host}/{fileId}"
 }
 
 type UploadResult struct {
@@ -153,7 +169,7 @@ func NewUploaderWithHttpClient(httpClient HTTPClient) *Uploader {
 	}
 }
 
-func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, data []byte) (fileId string, uploadResult *UploadResult, err error) {
+func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, data []byte) (fileId string, uploadResult *UploadResult, err error) {
 	doUploadFunc := func() error {
 		var host string
 		var auth security.EncodedJwt
@@ -162,7 +178,11 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho
 			return err
 		}
 
-		uploadOption.UploadUrl = genFileUrlFn(host, fileId)
+		genUrl := uploadOption.GenUploadUrl
+		if genUrl == nil {
+			genUrl = func(host, fileId string) string { return fmt.Sprintf("http://%s/%s", host, fileId) }
+		}
+		uploadOption.UploadUrl = genUrl(host, fileId)
 		uploadOption.Jwt = auth
 
 		uploadResult, err = uploader.retriedUploadData(context.Background(), data, uploadOption)
@@ -183,7 +203,7 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho
 
 // UploadWithRetry will retry both assigning volume request and uploading content
 // The option parameter does not need to specify UploadUrl and Jwt, which will come from assigning volume.
-func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assignRequest *filer_pb.AssignVolumeRequest, uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, reader io.Reader) (fileId string, uploadResult *UploadResult, err error, data []byte) {
+func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assignRequest *filer_pb.AssignVolumeRequest, uploadOption *UploadOption, reader io.Reader) (fileId string, uploadResult *UploadResult, err error, data []byte) {
 	bytesReader, ok := reader.(*util.BytesReader)
 	if ok {
 		data = bytesReader.Bytes
@@ -234,7 +254,7 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi
 			err = fmt.Errorf("filerGrpcAddress assign volume: %w", grpcAssignErr)
 		}
 		return
-	}, uploadOption, genFileUrlFn, data)
+	}, uploadOption, data)
 	return
 }
 

weed/operation/upload_content_test.go

@@ -107,9 +107,7 @@ func TestUploadWithRetryDataReassignsOnVolumeSizeExceeded(t *testing.T) {
 			return "1,first", "volume-a", "", nil
 		}
 		return "2,second", "volume-b", "", nil
-	}, &UploadOption{Filename: "test.bin"}, func(host, fileId string) string {
-		return "http://" + host + "/" + fileId
-	}, []byte("abc"))
+	}, &UploadOption{Filename: "test.bin"}, []byte("abc"))
 
 	if err != nil {
 		t.Fatalf("expected success after reassignment, got %v", err)
@@ -210,7 +208,6 @@ func TestUploadWithRetryBoundsAssignVolume(t *testing.T) {
 		_, _, err, _ := uploader.UploadWithRetry(client,
 			&filer_pb.AssignVolumeRequest{Count: 1},
 			&UploadOption{Filename: "test.bin"},
-			func(host, fileId string) string { return "http://" + host + "/" + fileId },
 			bytes.NewReader([]byte("abc")),
 		)
 		done <- err

weed/replication/sink/filersink/fetch_write.go

@@ -187,6 +187,17 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc
 
 	retryName := fmt.Sprintf("replicate manifest chunk %s", sourceFileId)
 	err = util.RetryUntil(retryName, func() error {
+		uploadOption := &operation.UploadOption{
+			Filename:          "",
+			Cipher:            false,
+			IsInputCompressed: false,
+			MimeType:          "application/octet-stream",
+			PairMap:           nil,
+			RetryForever:      false,
+		}
+		if fs.writeChunkByFiler {
+			uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address)
+		}
 		currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry(
 			fs,
 			&filer_pb.AssignVolumeRequest{
@@ -198,17 +209,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc
 				DiskType:    fs.diskType,
 				Path:        path,
 			},
-			&operation.UploadOption{
-				Filename:          "",
-				Cipher:            false,
-				IsInputCompressed: false,
-				MimeType:          "application/octet-stream",
-				PairMap:           nil,
-				RetryForever:      false,
-			},
-			func(host, fileId string) string {
-				return fs.buildUploadUrl(host, fileId)
-			},
+			uploadOption,
 			bytes.NewReader(manifestData),
 		)
 		if uploadErr != nil {
@@ -302,6 +303,18 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string,
 		transferStatus.Status = "uploading"
 		transferStatus.mu.Unlock()
 
+		uploadOption := &operation.UploadOption{
+			Filename:          savedFilename,
+			Cipher:            false,
+			IsInputCompressed: "gzip" == savedHeader.Get("Content-Encoding"),
+			MimeType:          savedHeader.Get("Content-Type"),
+			PairMap:           nil,
+			RetryForever:      false,
+			SourceUrl:         savedSourceUrl,
+		}
+		if fs.writeChunkByFiler {
+			uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address)
+		}
 		currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry(
 			fs,
 			&filer_pb.AssignVolumeRequest{
@@ -313,20 +326,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string,
 				DiskType:    fs.diskType,
 				Path:        path,
 			},
-			&operation.UploadOption{
-				Filename:          savedFilename,
-				Cipher:            false,
-				IsInputCompressed: "gzip" == savedHeader.Get("Content-Encoding"),
-				MimeType:          savedHeader.Get("Content-Type"),
-				PairMap:           nil,
-				RetryForever:      false,
-				SourceUrl:         savedSourceUrl,
-			},
-			func(host, fileId string) string {
-				fileUrl := fs.buildUploadUrl(host, fileId)
-				glog.V(4).Infof("replicating %s to %s header:%+v", savedFilename, fileUrl, savedHeader)
-				return fileUrl
-			},
+			uploadOption,
 			util.NewBytesReader(fullData),
 		)
 		if uploadErr != nil {
@@ -435,13 +435,6 @@ func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) e
 	return nil
 }
 
-func (fs *FilerSink) buildUploadUrl(host, fileId string) string {
-	if fs.writeChunkByFiler {
-		return fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)
-	}
-	return fmt.Sprintf("http://%s/%s", host, fileId)
-}
-
 // hasSourceNewerVersion reports whether the source's current entry for targetPath
 // has moved past sourceMtimeNs — gone, or a strictly-newer mtime — meaning the
 // version being replayed is stale. The lookup runs regardless of sourceMtimeNs so

weed/server/filer_server_handlers_read.go

@@ -254,5 +254,16 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 }
 
 func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string {
-	return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey(), fs.volumeGuard.ReadExpiresAfterSec(), fileId))
+	// Generate a read JWT for volume server access. If the dedicated
+	// read key (jwt.signing.read.key) is not configured, fall back to the
+	// general signing key (jwt.signing.key) so the proxy can still
+	// authenticate to volume servers that require JWT.
+	key := fs.volumeGuard.ReadSigningKey()
+	if len(key) == 0 {
+		key = fs.volumeGuard.SigningKey()
+	}
+	if len(key) == 0 {
+		return ""
+	}
+	return string(security.GenJwtForVolumeServer(key, fs.volumeGuard.ReadExpiresAfterSec(), fileId))
 }