s3: propagate storage rule fsync to volume server uploads (#10906)

The storage rule's fsync decision was computed by the filer
(detectStorageOption -> rule.Fsync) and applied on the filer's own HTTP
write path, but was never carried onto the chunk uploads S3 issues: the
AssignVolumeResponse had no fsync field, so the s3api client could not
learn the decision, and the chunked upload URL was hardcoded without it.
Every S3 write to a path with fsync configured went to the volume server
as a non-fsync write.

Carry the decision through the assign response:

- filer.proto: AssignVolumeResponse gains bool fsync, filled from the
  storage option the assign resolved.
- operation.AssignResult gains Fsync, so uploadChunk can append
  ?fsync=true to the volume server upload URL (single and replica
  fan-out paths).
- The S3 PUT/UploadPart assignFunc, the S3 copy path, the admin file
  browser upload, and the Iceberg worker assign functions all forward
  the response field.

Adds TestUploadReaderInChunksAppendsFsyncWhenAssigned.

weed/admin/handlers/file_browser_grpc.go

@@ -145,6 +145,7 @@ func (h *FileBrowserHandlers) uploadFileGrpc(ctx context.Context, filePath strin
 			PublicUrl: assignResp.Location.PublicUrl,
 			Count:     uint64(count),
 			Auth:      security.EncodedJwt(assignResp.Auth),
+			Fsync:     assignResp.Fsync,
 		}, nil
 	}
 

weed/operation/assign_file_id.go

@@ -38,6 +38,9 @@ type AssignResult struct {
 	Error     string              `json:"error,omitempty"`
 	Auth      security.EncodedJwt `json:"auth,omitempty"`
 	Replicas  []Location          `json:"replicas,omitempty"`
+	// Fsync carries the storage rule's fsync decision for the assigned path,
+	// so the upload request to the volume server can set ?fsync=true.
+	Fsync bool `json:"fsync,omitempty"`
 }
 
 func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest *VolumeAssignRequest, alternativeRequests ...*VolumeAssignRequest) (*AssignResult, error) {

weed/operation/upload_chunked.go

@@ -306,7 +306,7 @@ const chunkAssignAttempts = 3
 func uploadChunk(ctx context.Context, assignResult *AssignResult, data []byte, jwt security.EncodedJwt, md5b64 string, opt *ChunkedUploadOption) (*UploadResult, error) {
 	holders := chunkHolders(assignResult)
 	if opt.UploadFunc == nil && !opt.Cipher && len(holders) > 1 {
-		return uploadChunkToHolders(ctx, holders, assignResult.Fid, data, jwt, md5b64, opt)
+		return uploadChunkToHolders(ctx, holders, assignResult.Fid, data, jwt, md5b64, assignResult.Fsync, opt)
 	}
 	uploadOption := &UploadOption{
 		UploadUrl:         fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid),
@@ -325,6 +325,9 @@ func uploadChunk(ctx context.Context, assignResult *AssignResult, data []byte, j
 		// every holder and re-uploading the chunk.
 		MaxAttempts: 1,
 	}
+	if assignResult.Fsync {
+		uploadOption.UploadUrl += "?fsync=true"
+	}
 	// Use mock upload function if provided (for testing), otherwise use real uploader
 	if opt.UploadFunc != nil {
 		return opt.UploadFunc(ctx, data, uploadOption)
@@ -351,7 +354,7 @@ func chunkHolders(assignResult *AssignResult) []string {
 // type=replicate). On the first failure it cancels the remaining uploads and
 // deletes any copies that already landed, so a partial fan-out leaves no
 // orphaned needle the caller cannot see.
-func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data []byte, jwt security.EncodedJwt, md5b64 string, opt *ChunkedUploadOption) (*UploadResult, error) {
+func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data []byte, jwt security.EncodedJwt, md5b64 string, fsync bool, opt *ChunkedUploadOption) (*UploadResult, error) {
 	uploader, err := NewUploader()
 	if err != nil {
 		return nil, fmt.Errorf("create uploader: %w", err)
@@ -367,8 +370,12 @@ func uploadChunkToHolders(ctx context.Context, hosts []string, fid string, data
 	outcomes := make(chan outcome, len(hosts))
 	for _, host := range hosts {
 		go func(host string) {
+			uploadUrl := fmt.Sprintf("http://%s/%s?type=replicate", host, fid)
+			if fsync {
+				uploadUrl += "&fsync=true"
+			}
 			uploadOption := &UploadOption{
-				UploadUrl:         fmt.Sprintf("http://%s/%s?type=replicate", host, fid),
+				UploadUrl:         uploadUrl,
 				Cipher:            false,
 				IsInputCompressed: false,
 				MimeType:          opt.MimeType,

weed/operation/upload_chunked_test.go

@@ -280,7 +280,7 @@ func TestUploadChunkToHoldersRollsBackOnPartialFailure(t *testing.T) {
 	defer bad.Close()
 
 	hosts := []string{strings.TrimPrefix(good.URL, "http://"), strings.TrimPrefix(bad.URL, "http://")}
-	_, err := uploadChunkToHolders(context.Background(), hosts, fid, []byte("hello world"), "", "", &ChunkedUploadOption{})
+	_, err := uploadChunkToHolders(context.Background(), hosts, fid, []byte("hello world"), "", "", false, &ChunkedUploadOption{})
 
 	if err == nil {
 		t.Fatal("expected error from a partial fan-out")
@@ -639,3 +639,43 @@ func TestUploadReaderInChunksDoesNotMultiplyRelayAttempts(t *testing.T) {
 		t.Errorf("expected every abandoned fid to be rolled back, got %d deletes", got)
 	}
 }
+
+// TestUploadReaderInChunksAppendsFsyncWhenAssigned verifies that a chunk
+// assignment carrying the storage rule's fsync decision lands as ?fsync=true
+// on the volume server upload URL.
+func TestUploadReaderInChunksAppendsFsyncWhenAssigned(t *testing.T) {
+	testData := []byte("data needing fsync")
+	reader := bytes.NewReader(testData)
+
+	assignFunc := func(ctx context.Context, count int, expectedDataSize uint64) (*VolumeAssignRequest, *AssignResult, error) {
+		return nil, &AssignResult{
+			Fid:       "test-fid,1234",
+			Url:       "http://test-volume:8080",
+			PublicUrl: "http://test-volume:8080",
+			Count:     1,
+			Fsync:     true,
+		}, nil
+	}
+
+	var gotUploadUrl string
+	uploadFunc := func(ctx context.Context, data []byte, option *UploadOption) (*UploadResult, error) {
+		gotUploadUrl = option.UploadUrl
+		return &UploadResult{
+			Size:       uint32(len(data)),
+			ContentMd5: "mock-md5-hash",
+		}, nil
+	}
+
+	if _, err := UploadReaderInChunks(context.Background(), reader, &ChunkedUploadOption{
+		ChunkSize:       8 * 1024,
+		SmallFileLimit:  256,
+		SaveSmallInline: false,
+		AssignFunc:      assignFunc,
+		UploadFunc:      uploadFunc,
+	}); err != nil {
+		t.Fatalf("Expected successful upload, got error: %v", err)
+	}
+	if !strings.Contains(gotUploadUrl, "?fsync=true") {
+		t.Errorf("Expected the upload URL to carry ?fsync=true when the assignment says fsync, got %q", gotUploadUrl)
+	}
+}

weed/pb/filer.proto

@@ -561,6 +561,9 @@ message AssignVolumeResponse {
     string error = 8;
     Location location = 9;
     repeated Location replicas = 10;
+    // fsync is the storage rule's fsync decision for the assigned path, so the
+    // client can carry it onto the volume server upload request.
+    bool fsync = 11;
 }
 
 message LookupVolumeRequest {

weed/pb/filer_pb/filer.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.6
-// 	protoc        v7.35.0
+// 	protoc        v7.35.1
 // source: filer.proto
 
 package filer_pb
@@ -3230,15 +3230,18 @@ func (x *AssignVolumeRequest) GetExpectedDataSize() uint64 {
 }
 
 type AssignVolumeResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	FileId        string                 `protobuf:"bytes,1,opt,name=file_id,json=fileId,proto3" json:"file_id,omitempty"`
-	Count         int32                  `protobuf:"varint,4,opt,name=count,proto3" json:"count,omitempty"`
-	Auth          string                 `protobuf:"bytes,5,opt,name=auth,proto3" json:"auth,omitempty"`
-	Collection    string                 `protobuf:"bytes,6,opt,name=collection,proto3" json:"collection,omitempty"`
-	Replication   string                 `protobuf:"bytes,7,opt,name=replication,proto3" json:"replication,omitempty"`
-	Error         string                 `protobuf:"bytes,8,opt,name=error,proto3" json:"error,omitempty"`
-	Location      *Location              `protobuf:"bytes,9,opt,name=location,proto3" json:"location,omitempty"`
-	Replicas      []*Location            `protobuf:"bytes,10,rep,name=replicas,proto3" json:"replicas,omitempty"`
+	state       protoimpl.MessageState `protogen:"open.v1"`
+	FileId      string                 `protobuf:"bytes,1,opt,name=file_id,json=fileId,proto3" json:"file_id,omitempty"`
+	Count       int32                  `protobuf:"varint,4,opt,name=count,proto3" json:"count,omitempty"`
+	Auth        string                 `protobuf:"bytes,5,opt,name=auth,proto3" json:"auth,omitempty"`
+	Collection  string                 `protobuf:"bytes,6,opt,name=collection,proto3" json:"collection,omitempty"`
+	Replication string                 `protobuf:"bytes,7,opt,name=replication,proto3" json:"replication,omitempty"`
+	Error       string                 `protobuf:"bytes,8,opt,name=error,proto3" json:"error,omitempty"`
+	Location    *Location              `protobuf:"bytes,9,opt,name=location,proto3" json:"location,omitempty"`
+	Replicas    []*Location            `protobuf:"bytes,10,rep,name=replicas,proto3" json:"replicas,omitempty"`
+	// fsync is the storage rule's fsync decision for the assigned path, so the
+	// client can carry it onto the volume server upload request.
+	Fsync         bool `protobuf:"varint,11,opt,name=fsync,proto3" json:"fsync,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -3329,6 +3332,13 @@ func (x *AssignVolumeResponse) GetReplicas() []*Location {
 	return nil
 }
 
+func (x *AssignVolumeResponse) GetFsync() bool {
+	if x != nil {
+		return x.Fsync
+	}
+	return false
+}
+
 type LookupVolumeRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	VolumeIds     []string               `protobuf:"bytes,1,rep,name=volume_ids,json=volumeIds,proto3" json:"volume_ids,omitempty"`
@@ -7259,7 +7269,7 @@ const file_filer_proto_rawDesc = "" +
 	"\tdata_node\x18\t \x01(\tR\bdataNode\x12\x1b\n" +
 	"\tdisk_type\x18\b \x01(\tR\bdiskType\x12,\n" +
 	"\x12expected_data_size\x18\n" +
-	" \x01(\x04R\x10expectedDataSize\"\x91\x02\n" +
+	" \x01(\x04R\x10expectedDataSize\"\xa7\x02\n" +
 	"\x14AssignVolumeResponse\x12\x17\n" +
 	"\afile_id\x18\x01 \x01(\tR\x06fileId\x12\x14\n" +
 	"\x05count\x18\x04 \x01(\x05R\x05count\x12\x12\n" +
@@ -7271,7 +7281,8 @@ const file_filer_proto_rawDesc = "" +
 	"\x05error\x18\b \x01(\tR\x05error\x12.\n" +
 	"\blocation\x18\t \x01(\v2\x12.filer_pb.LocationR\blocation\x12.\n" +
 	"\breplicas\x18\n" +
-	" \x03(\v2\x12.filer_pb.LocationR\breplicas\"4\n" +
+	" \x03(\v2\x12.filer_pb.LocationR\breplicas\x12\x14\n" +
+	"\x05fsync\x18\v \x01(\bR\x05fsync\"4\n" +
 	"\x13LookupVolumeRequest\x12\x1d\n" +
 	"\n" +
 	"volume_ids\x18\x01 \x03(\tR\tvolumeIds\"=\n" +

weed/pb/filer_pb/filer_grpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.6.2
-// - protoc             v7.35.0
+// - protoc             v7.35.1
 // source: filer.proto
 
 package filer_pb

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -2944,6 +2944,16 @@ func (m *AssignVolumeResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error)
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.Fsync {
+		i--
+		if m.Fsync {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x58
+	}
 	if len(m.Replicas) > 0 {
 		for iNdEx := len(m.Replicas) - 1; iNdEx >= 0; iNdEx-- {
 			size, err := m.Replicas[iNdEx].MarshalToSizedBufferVT(dAtA[:i])
@@ -7424,6 +7434,9 @@ func (m *AssignVolumeResponse) SizeVT() (n int) {
 			n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 		}
 	}
+	if m.Fsync {
+		n += 2
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -17003,6 +17016,26 @@ func (m *AssignVolumeResponse) UnmarshalVT(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 11:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Fsync", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Fsync = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])

weed/s3api/s3api_object_handlers_copy.go

@@ -1706,6 +1706,9 @@ const multipartFramingOverhead = 1024
 // buffer is GC'd as soon as the upload returns.
 func newChunkUploadOption(chunkData []byte, assignResult *filer_pb.AssignVolumeResponse, isCompressed bool) *operation.UploadOption {
 	dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId)
+	if assignResult.Fsync {
+		dstUrl += "?fsync=true"
+	}
 	return &operation.UploadOption{
 		UploadUrl:         dstUrl,
 		Cipher:            false, // Data is already encrypted if source had CipherKey; don't re-encrypt

weed/s3api/s3api_object_handlers_put.go

@@ -550,6 +550,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 			PublicUrl: assignResult.Location.PublicUrl,
 			Count:     uint64(count),
 			Auth:      security.EncodedJwt(assignResult.Auth),
+			Fsync:     assignResult.Fsync,
 		}
 		for _, replica := range assignResult.Replicas {
 			result.Replicas = append(result.Replicas, operation.Location{

weed/server/filer_grpc_server.go

@@ -859,6 +859,7 @@ func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVol
 		Auth:        string(assignResult.Auth),
 		Collection:  so.Collection,
 		Replication: so.Replication,
+		Fsync:       so.Fsync,
 	}
 	// Forward the replica holders so a client can write all copies directly.
 	for _, replica := range assignResult.Replicas {

weed/worker/tasks/iceberg/filer_io.go

@@ -336,6 +336,7 @@ func uploadFilerChunks(ctx context.Context, client filer_pb.SeaweedFilerClient,
 			PublicUrl: resp.Location.PublicUrl,
 			Count:     uint64(count),
 			Auth:      security.EncodedJwt(resp.Auth),
+			Fsync:     resp.Fsync,
 		}, nil
 	}