From c80664ec21a5e4217e0255e46f6cd315e500a1d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AD=99=E8=B6=85?= Date: Mon, 24 Aug 2026 13:11:08 +0800 Subject: [PATCH] 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 | 1 + weed/operation/assign_file_id.go | 3 ++ weed/operation/upload_chunked.go | 13 ++++++-- weed/operation/upload_chunked_test.go | 42 +++++++++++++++++++++++- weed/pb/filer.proto | 3 ++ weed/pb/filer_pb/filer.pb.go | 35 +++++++++++++------- weed/pb/filer_pb/filer_grpc.pb.go | 2 +- weed/pb/filer_pb/filer_vtproto.pb.go | 33 +++++++++++++++++++ weed/s3api/s3api_object_handlers_copy.go | 3 ++ weed/s3api/s3api_object_handlers_put.go | 1 + weed/server/filer_grpc_server.go | 1 + weed/worker/tasks/iceberg/filer_io.go | 1 + 12 files changed, 121 insertions(+), 17 deletions(-) diff --git a/weed/admin/handlers/file_browser_grpc.go b/weed/admin/handlers/file_browser_grpc.go index a79e84d7e..0b2f9eb98 100644 --- a/weed/admin/handlers/file_browser_grpc.go +++ b/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 } diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index 8ae2049b3..bddc1ae9c 100644 --- a/weed/operation/assign_file_id.go +++ b/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) { diff --git a/weed/operation/upload_chunked.go b/weed/operation/upload_chunked.go index db8dec89c..888ce6427 100644 --- a/weed/operation/upload_chunked.go +++ b/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, diff --git a/weed/operation/upload_chunked_test.go b/weed/operation/upload_chunked_test.go index 1f26256a1..b1accb3b7 100644 --- a/weed/operation/upload_chunked_test.go +++ b/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) + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index c6f00af3f..adef26743 100644 --- a/weed/pb/filer.proto +++ b/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 { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 6e7f65a72..b95a2cfe5 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/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" + diff --git a/weed/pb/filer_pb/filer_grpc.pb.go b/weed/pb/filer_pb/filer_grpc.pb.go index 0d103b4c5..2bbe9e23c 100644 --- a/weed/pb/filer_pb/filer_grpc.pb.go +++ b/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 diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 1f6b2ea0f..8ba5613e0 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/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:]) diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 01b764d0c..a704e8957 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/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 diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 3ecb83c35..e83fd35b1 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/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{ diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 0e78b7046..07922f2ed 100644 --- a/weed/server/filer_grpc_server.go +++ b/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 { diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 93d914c70..a1c8e1c5b 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/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 }