diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index feb37fbd7f..2a21f1e947 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -359,6 +359,8 @@ message ObjectMutation { bool touch_mtime = 12; // PATCH_EXTENDED: set the entry's Mtime to now (e.g. a metadata-replace copy) bool remove_empty_parent = 13; // DELETE: also remove the parent directory when the delete leaves it empty (best-effort) bool keep_remote_object = 14; // DELETE: leave the object on the mounted remote storage + WriteCondition condition = 15; // optional precondition checked under the transaction lock just before this mutation applies + string condition_key = 16; // entry the condition evaluates against; empty = the mutation's target } // Recompute re-derives a pointer entry (directory/name on the mutation) from the @@ -404,6 +406,7 @@ message ObjectTransactionRequest { message ObjectTransactionResponse { string error = 1; FilerError error_code = 2; + optional int32 failed_mutation = 3; // index of the mutation that failed (its condition or its apply); unset when the request-level condition failed } // PosixLockRange is one advisory byte-range lock. Owner identity is (sid, owner): diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index feb37fbd7f..dd1d6edb05 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -359,6 +359,8 @@ message ObjectMutation { bool touch_mtime = 12; // PATCH_EXTENDED: set the entry's Mtime to now (e.g. a metadata-replace copy) bool remove_empty_parent = 13; // DELETE: also remove the parent directory when the delete leaves it empty (best-effort) bool keep_remote_object = 14; // DELETE: leave the object on the mounted remote storage + WriteCondition condition = 15; // optional precondition checked under the transaction lock just before this mutation applies + string condition_key = 16; // entry the condition evaluates against; empty = the mutation's target } // Recompute re-derives a pointer entry (directory/name on the mutation) from the @@ -399,11 +401,13 @@ message ObjectTransactionRequest { bool is_moved = 8; // set on a forwarded transaction so the receiver applies it locally instead of forwarding again string request_id = 9; // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying string condition_alt_key = 10; // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) + bool condition_prefer_alt = 11; // evaluate condition_alt_key (the latest version) first; fall back to the primary entry only when the alt path cannot resolve } message ObjectTransactionResponse { string error = 1; FilerError error_code = 2; + optional int32 failed_mutation = 3; // index of the mutation that failed (its condition or its apply); unset when the request-level condition failed } // PosixLockRange is one advisory byte-range lock. Owner identity is (sid, owner): diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index cb5f44b2b5..e4efbf7528 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1561,6 +1561,8 @@ type ObjectMutation struct { TouchMtime bool `protobuf:"varint,12,opt,name=touch_mtime,json=touchMtime,proto3" json:"touch_mtime,omitempty"` // PATCH_EXTENDED: set the entry's Mtime to now (e.g. a metadata-replace copy) RemoveEmptyParent bool `protobuf:"varint,13,opt,name=remove_empty_parent,json=removeEmptyParent,proto3" json:"remove_empty_parent,omitempty"` // DELETE: also remove the parent directory when the delete leaves it empty (best-effort) KeepRemoteObject bool `protobuf:"varint,14,opt,name=keep_remote_object,json=keepRemoteObject,proto3" json:"keep_remote_object,omitempty"` // DELETE: leave the object on the mounted remote storage + Condition *WriteCondition `protobuf:"bytes,15,opt,name=condition,proto3" json:"condition,omitempty"` // optional precondition checked under the transaction lock just before this mutation applies + ConditionKey string `protobuf:"bytes,16,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // entry the condition evaluates against; empty = the mutation's target unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1693,6 +1695,20 @@ func (x *ObjectMutation) GetKeepRemoteObject() bool { return false } +func (x *ObjectMutation) GetCondition() *WriteCondition { + if x != nil { + return x.Condition + } + return nil +} + +func (x *ObjectMutation) GetConditionKey() string { + if x != nil { + return x.ConditionKey + } + return "" +} + // Recompute re-derives a pointer entry (directory/name on the mutation) from the // current contents of a scanned directory, atomically under the transaction's // lock. It is mechanical: the filer picks the child that sorts first or last by @@ -1830,11 +1846,12 @@ type ObjectTransactionRequest struct { Mutations []*ObjectMutation `protobuf:"bytes,3,rep,name=mutations,proto3" json:"mutations,omitempty"` IsFromOtherCluster bool `protobuf:"varint,4,opt,name=is_from_other_cluster,json=isFromOtherCluster,proto3" json:"is_from_other_cluster,omitempty"` Signatures []int32 `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"` - ConditionKey string `protobuf:"bytes,6,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key) - RouteKey string `protobuf:"bytes,7,opt,name=route_key,json=routeKey,proto3" json:"route_key,omitempty"` // ring key identifying the owner filer; a non-owner forwards the whole transaction to it - IsMoved bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded transaction so the receiver applies it locally instead of forwarding again - RequestId string `protobuf:"bytes,9,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying - ConditionAltKey string `protobuf:"bytes,10,opt,name=condition_alt_key,json=conditionAltKey,proto3" json:"condition_alt_key,omitempty"` // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) + ConditionKey string `protobuf:"bytes,6,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key) + RouteKey string `protobuf:"bytes,7,opt,name=route_key,json=routeKey,proto3" json:"route_key,omitempty"` // ring key identifying the owner filer; a non-owner forwards the whole transaction to it + IsMoved bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded transaction so the receiver applies it locally instead of forwarding again + RequestId string `protobuf:"bytes,9,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying + ConditionAltKey string `protobuf:"bytes,10,opt,name=condition_alt_key,json=conditionAltKey,proto3" json:"condition_alt_key,omitempty"` // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) + ConditionPreferAlt bool `protobuf:"varint,11,opt,name=condition_prefer_alt,json=conditionPreferAlt,proto3" json:"condition_prefer_alt,omitempty"` // evaluate condition_alt_key (the latest version) first; fall back to the primary entry only when the alt path cannot resolve unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1939,12 +1956,20 @@ func (x *ObjectTransactionRequest) GetConditionAltKey() string { return "" } +func (x *ObjectTransactionRequest) GetConditionPreferAlt() bool { + if x != nil { + return x.ConditionPreferAlt + } + return false +} + type ObjectTransactionResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` - ErrorCode FilerError `protobuf:"varint,2,opt,name=error_code,json=errorCode,proto3,enum=filer_pb.FilerError" json:"error_code,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` + ErrorCode FilerError `protobuf:"varint,2,opt,name=error_code,json=errorCode,proto3,enum=filer_pb.FilerError" json:"error_code,omitempty"` + FailedMutation *int32 `protobuf:"varint,3,opt,name=failed_mutation,json=failedMutation,proto3,oneof" json:"failed_mutation,omitempty"` // index of the mutation that failed (its condition or its apply); unset when the request-level condition failed + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ObjectTransactionResponse) Reset() { @@ -1991,6 +2016,13 @@ func (x *ObjectTransactionResponse) GetErrorCode() FilerError { return FilerError_OK } +func (x *ObjectTransactionResponse) GetFailedMutation() int32 { + if x != nil && x.FailedMutation != nil { + return *x.FailedMutation + } + return 0 +} + // PosixLockRange is one advisory byte-range lock. Owner identity is (sid, owner): // sid is the mount session, owner the FUSE lock owner within it, so owners from // different mounts never alias. end is inclusive (max uint64 = to EOF); is_flock @@ -7497,7 +7529,7 @@ const file_filer_proto_rawDesc = "" + "\x18IF_EXTENDED_TIME_ELAPSED\x10\b\x12\x13\n" + "\x0fIF_CHUNKS_EQUAL\x10\t\x12\x12\n" + "\x0eIF_ENTRY_EQUAL\x10\n" + - "\"\xd0\x05\n" + + "\"\xad\x06\n" + "\x0eObjectMutation\x121\n" + "\x04type\x18\x01 \x01(\x0e2\x1d.filer_pb.ObjectMutation.TypeR\x04type\x12\x1c\n" + "\tdirectory\x18\x02 \x01(\tR\tdirectory\x12\x12\n" + @@ -7515,7 +7547,9 @@ const file_filer_proto_rawDesc = "" + "\vtouch_mtime\x18\f \x01(\bR\n" + "touchMtime\x12.\n" + "\x13remove_empty_parent\x18\r \x01(\bR\x11removeEmptyParent\x12,\n" + - "\x12keep_remote_object\x18\x0e \x01(\bR\x10keepRemoteObject\x1a>\n" + + "\x12keep_remote_object\x18\x0e \x01(\bR\x10keepRemoteObject\x126\n" + + "\tcondition\x18\x0f \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12#\n" + + "\rcondition_key\x18\x10 \x01(\tR\fconditionKey\x1a>\n" + "\x10SetExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"E\n" + @@ -7543,7 +7577,7 @@ const file_filer_proto_rawDesc = "" + " \x01(\tR\x11onlyIfCurrentName\x1a?\n" + "\x11CopyExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xa0\x03\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xd2\x03\n" + "\x18ObjectTransactionRequest\x12\x19\n" + "\block_key\x18\x01 \x01(\tR\alockKey\x126\n" + "\tcondition\x18\x02 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x126\n" + @@ -7558,11 +7592,14 @@ const file_filer_proto_rawDesc = "" + "\n" + "request_id\x18\t \x01(\tR\trequestId\x12*\n" + "\x11condition_alt_key\x18\n" + - " \x01(\tR\x0fconditionAltKey\"f\n" + + " \x01(\tR\x0fconditionAltKey\x120\n" + + "\x14condition_prefer_alt\x18\v \x01(\bR\x12conditionPreferAlt\"\xa8\x01\n" + "\x19ObjectTransactionResponse\x12\x14\n" + "\x05error\x18\x01 \x01(\tR\x05error\x123\n" + "\n" + - "error_code\x18\x02 \x01(\x0e2\x14.filer_pb.FilerErrorR\terrorCode\"\xa1\x01\n" + + "error_code\x18\x02 \x01(\x0e2\x14.filer_pb.FilerErrorR\terrorCode\x12,\n" + + "\x0ffailed_mutation\x18\x03 \x01(\x05H\x00R\x0efailedMutation\x88\x01\x01B\x12\n" + + "\x10_failed_mutation\"\xa1\x01\n" + "\x0ePosixLockRange\x12\x14\n" + "\x05start\x18\x01 \x01(\x04R\x05start\x12\x10\n" + "\x03end\x18\x02 \x01(\x04R\x03end\x12\x12\n" + @@ -8220,136 +8257,137 @@ var file_filer_proto_depIdxs = []int32{ 10, // 17: filer_pb.ObjectMutation.entry:type_name -> filer_pb.Entry 101, // 18: filer_pb.ObjectMutation.set_extended:type_name -> filer_pb.ObjectMutation.SetExtendedEntry 20, // 19: filer_pb.ObjectMutation.recompute:type_name -> filer_pb.Recompute - 102, // 20: filer_pb.Recompute.copy_extended:type_name -> filer_pb.Recompute.CopyExtendedEntry - 18, // 21: filer_pb.ObjectTransactionRequest.condition:type_name -> filer_pb.WriteCondition - 19, // 22: filer_pb.ObjectTransactionRequest.mutations:type_name -> filer_pb.ObjectMutation - 1, // 23: filer_pb.ObjectTransactionResponse.error_code:type_name -> filer_pb.FilerError - 2, // 24: filer_pb.PosixLockRequest.op:type_name -> filer_pb.PosixLockOp - 23, // 25: filer_pb.PosixLockRequest.lock:type_name -> filer_pb.PosixLockRange - 23, // 26: filer_pb.PosixLockRequest.locks:type_name -> filer_pb.PosixLockRange - 23, // 27: filer_pb.PosixLockResponse.conflict:type_name -> filer_pb.PosixLockRange - 21, // 28: filer_pb.ObjectTransactionBatchRequest.transactions:type_name -> filer_pb.ObjectTransactionRequest - 22, // 29: filer_pb.ObjectTransactionBatchResponse.responses:type_name -> filer_pb.ObjectTransactionResponse - 59, // 30: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 1, // 31: filer_pb.CreateEntryResponse.error_code:type_name -> filer_pb.FilerError - 10, // 32: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry - 103, // 33: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry - 18, // 34: filer_pb.UpdateEntryRequest.condition:type_name -> filer_pb.WriteCondition - 59, // 35: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 13, // 36: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk - 104, // 37: filer_pb.DeleteEntryRequest.expected_extended:type_name -> filer_pb.DeleteEntryRequest.ExpectedExtendedEntry - 59, // 38: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 12, // 39: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification - 45, // 40: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location - 45, // 41: filer_pb.AssignVolumeResponse.replicas:type_name -> filer_pb.Location - 1, // 42: filer_pb.AssignVolumeResponse.error_code:type_name -> filer_pb.FilerError - 45, // 43: filer_pb.Locations.locations:type_name -> filer_pb.Location - 105, // 44: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry - 47, // 45: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection - 12, // 46: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification - 59, // 47: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse - 63, // 48: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef - 62, // 49: filer_pb.ListMetadataSubscribersResponse.subscribers:type_name -> filer_pb.MetadataSubscriber - 13, // 50: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk - 10, // 51: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry - 106, // 52: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource - 107, // 53: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf - 10, // 54: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry - 59, // 55: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse - 84, // 56: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock - 17, // 57: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest - 29, // 58: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest - 35, // 59: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest - 39, // 60: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest - 28, // 61: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse - 30, // 62: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse - 36, // 63: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse - 40, // 64: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse - 95, // 65: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo - 5, // 66: filer_pb.LookupDirectoryEntriesRequest.requests:type_name -> filer_pb.LookupDirectoryEntryRequest - 10, // 67: filer_pb.LookupDirectoryEntryResult.entry:type_name -> filer_pb.Entry - 97, // 68: filer_pb.LookupDirectoryEntriesResponse.results:type_name -> filer_pb.LookupDirectoryEntryResult - 108, // 69: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry - 109, // 70: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry - 3, // 71: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind - 10, // 72: filer_pb.WriteCondition.Clause.expected_entry:type_name -> filer_pb.Entry - 44, // 73: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 44, // 74: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations - 5, // 75: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest - 96, // 76: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest - 7, // 77: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest - 17, // 78: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest - 29, // 79: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest - 31, // 80: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest - 33, // 81: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest - 35, // 82: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest - 21, // 83: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest - 26, // 84: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest - 24, // 85: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest - 37, // 86: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest - 39, // 87: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest - 89, // 88: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest - 41, // 89: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest - 43, // 90: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest - 48, // 91: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest - 50, // 92: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest - 52, // 93: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest - 54, // 94: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest - 56, // 95: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest - 64, // 96: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest - 58, // 97: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 58, // 98: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest - 60, // 99: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest - 71, // 100: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest - 73, // 101: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest - 76, // 102: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest - 78, // 103: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest - 80, // 104: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest - 82, // 105: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest - 85, // 106: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest - 87, // 107: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest - 91, // 108: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest - 93, // 109: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest - 6, // 110: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse - 98, // 111: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse - 8, // 112: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse - 28, // 113: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse - 30, // 114: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse - 32, // 115: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse - 34, // 116: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse - 36, // 117: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse - 22, // 118: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse - 27, // 119: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse - 25, // 120: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse - 38, // 121: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse - 40, // 122: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse - 90, // 123: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse - 42, // 124: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse - 46, // 125: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse - 49, // 126: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse - 51, // 127: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse - 53, // 128: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse - 55, // 129: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse - 57, // 130: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse - 65, // 131: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse - 59, // 132: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 59, // 133: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse - 61, // 134: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse - 72, // 135: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse - 74, // 136: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse - 77, // 137: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse - 79, // 138: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse - 81, // 139: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse - 83, // 140: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse - 86, // 141: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse - 88, // 142: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse - 92, // 143: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse - 94, // 144: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse - 110, // [110:145] is the sub-list for method output_type - 75, // [75:110] is the sub-list for method input_type - 75, // [75:75] is the sub-list for extension type_name - 75, // [75:75] is the sub-list for extension extendee - 0, // [0:75] is the sub-list for field type_name + 18, // 20: filer_pb.ObjectMutation.condition:type_name -> filer_pb.WriteCondition + 102, // 21: filer_pb.Recompute.copy_extended:type_name -> filer_pb.Recompute.CopyExtendedEntry + 18, // 22: filer_pb.ObjectTransactionRequest.condition:type_name -> filer_pb.WriteCondition + 19, // 23: filer_pb.ObjectTransactionRequest.mutations:type_name -> filer_pb.ObjectMutation + 1, // 24: filer_pb.ObjectTransactionResponse.error_code:type_name -> filer_pb.FilerError + 2, // 25: filer_pb.PosixLockRequest.op:type_name -> filer_pb.PosixLockOp + 23, // 26: filer_pb.PosixLockRequest.lock:type_name -> filer_pb.PosixLockRange + 23, // 27: filer_pb.PosixLockRequest.locks:type_name -> filer_pb.PosixLockRange + 23, // 28: filer_pb.PosixLockResponse.conflict:type_name -> filer_pb.PosixLockRange + 21, // 29: filer_pb.ObjectTransactionBatchRequest.transactions:type_name -> filer_pb.ObjectTransactionRequest + 22, // 30: filer_pb.ObjectTransactionBatchResponse.responses:type_name -> filer_pb.ObjectTransactionResponse + 59, // 31: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 1, // 32: filer_pb.CreateEntryResponse.error_code:type_name -> filer_pb.FilerError + 10, // 33: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry + 103, // 34: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry + 18, // 35: filer_pb.UpdateEntryRequest.condition:type_name -> filer_pb.WriteCondition + 59, // 36: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 13, // 37: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk + 104, // 38: filer_pb.DeleteEntryRequest.expected_extended:type_name -> filer_pb.DeleteEntryRequest.ExpectedExtendedEntry + 59, // 39: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 12, // 40: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification + 45, // 41: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location + 45, // 42: filer_pb.AssignVolumeResponse.replicas:type_name -> filer_pb.Location + 1, // 43: filer_pb.AssignVolumeResponse.error_code:type_name -> filer_pb.FilerError + 45, // 44: filer_pb.Locations.locations:type_name -> filer_pb.Location + 105, // 45: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry + 47, // 46: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection + 12, // 47: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification + 59, // 48: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse + 63, // 49: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef + 62, // 50: filer_pb.ListMetadataSubscribersResponse.subscribers:type_name -> filer_pb.MetadataSubscriber + 13, // 51: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk + 10, // 52: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry + 106, // 53: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource + 107, // 54: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf + 10, // 55: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry + 59, // 56: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse + 84, // 57: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock + 17, // 58: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest + 29, // 59: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest + 35, // 60: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest + 39, // 61: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest + 28, // 62: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse + 30, // 63: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse + 36, // 64: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse + 40, // 65: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse + 95, // 66: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo + 5, // 67: filer_pb.LookupDirectoryEntriesRequest.requests:type_name -> filer_pb.LookupDirectoryEntryRequest + 10, // 68: filer_pb.LookupDirectoryEntryResult.entry:type_name -> filer_pb.Entry + 97, // 69: filer_pb.LookupDirectoryEntriesResponse.results:type_name -> filer_pb.LookupDirectoryEntryResult + 108, // 70: filer_pb.LookupDirectoryEntriesResponse.locations_map:type_name -> filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry + 109, // 71: filer_pb.LookupDirectoryEntriesResponse.read_auth:type_name -> filer_pb.LookupDirectoryEntriesResponse.ReadAuthEntry + 3, // 72: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind + 10, // 73: filer_pb.WriteCondition.Clause.expected_entry:type_name -> filer_pb.Entry + 44, // 74: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 44, // 75: filer_pb.LookupDirectoryEntriesResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations + 5, // 76: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest + 96, // 77: filer_pb.SeaweedFiler.LookupDirectoryEntries:input_type -> filer_pb.LookupDirectoryEntriesRequest + 7, // 78: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest + 17, // 79: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest + 29, // 80: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest + 31, // 81: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest + 33, // 82: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest + 35, // 83: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest + 21, // 84: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest + 26, // 85: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest + 24, // 86: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest + 37, // 87: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest + 39, // 88: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest + 89, // 89: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest + 41, // 90: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest + 43, // 91: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest + 48, // 92: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest + 50, // 93: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest + 52, // 94: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest + 54, // 95: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest + 56, // 96: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest + 64, // 97: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest + 58, // 98: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 58, // 99: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest + 60, // 100: filer_pb.SeaweedFiler.ListMetadataSubscribers:input_type -> filer_pb.ListMetadataSubscribersRequest + 71, // 101: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest + 73, // 102: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest + 76, // 103: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest + 78, // 104: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest + 80, // 105: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest + 82, // 106: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest + 85, // 107: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest + 87, // 108: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest + 91, // 109: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest + 93, // 110: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest + 6, // 111: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse + 98, // 112: filer_pb.SeaweedFiler.LookupDirectoryEntries:output_type -> filer_pb.LookupDirectoryEntriesResponse + 8, // 113: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse + 28, // 114: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse + 30, // 115: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse + 32, // 116: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse + 34, // 117: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse + 36, // 118: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse + 22, // 119: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse + 27, // 120: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse + 25, // 121: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse + 38, // 122: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse + 40, // 123: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse + 90, // 124: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse + 42, // 125: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse + 46, // 126: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse + 49, // 127: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse + 51, // 128: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse + 53, // 129: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse + 55, // 130: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse + 57, // 131: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse + 65, // 132: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse + 59, // 133: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 59, // 134: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse + 61, // 135: filer_pb.SeaweedFiler.ListMetadataSubscribers:output_type -> filer_pb.ListMetadataSubscribersResponse + 72, // 136: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse + 74, // 137: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse + 77, // 138: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse + 79, // 139: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse + 81, // 140: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse + 83, // 141: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse + 86, // 142: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse + 88, // 143: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse + 92, // 144: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse + 94, // 145: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse + 111, // [111:146] is the sub-list for method output_type + 76, // [76:111] is the sub-list for method input_type + 76, // [76:76] is the sub-list for extension type_name + 76, // [76:76] is the sub-list for extension extendee + 0, // [0:76] is the sub-list for field type_name } func init() { file_filer_proto_init() } @@ -8358,6 +8396,7 @@ func file_filer_proto_init() { return } file_filer_proto_msgTypes[4].OneofWrappers = []any{} + file_filer_proto_msgTypes[17].OneofWrappers = []any{} file_filer_proto_msgTypes[84].OneofWrappers = []any{ (*StreamMutateEntryRequest_CreateRequest)(nil), (*StreamMutateEntryRequest_UpdateRequest)(nil), diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 8cd7734a72..9a3b40547c 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1294,6 +1294,25 @@ func (m *ObjectMutation) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.ConditionKey) > 0 { + i -= len(m.ConditionKey) + copy(dAtA[i:], m.ConditionKey) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.ConditionKey))) + i-- + dAtA[i] = 0x1 + i-- + dAtA[i] = 0x82 + } + if m.Condition != nil { + size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = protohelpers.EncodeVarint(dAtA, i, uint64(size)) + i-- + dAtA[i] = 0x7a + } if m.KeepRemoteObject { i-- if m.KeepRemoteObject { @@ -1579,6 +1598,16 @@ func (m *ObjectTransactionRequest) MarshalToSizedBufferVT(dAtA []byte) (int, err i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.ConditionPreferAlt { + i-- + if m.ConditionPreferAlt { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x58 + } if len(m.ConditionAltKey) > 0 { i -= len(m.ConditionAltKey) copy(dAtA[i:], m.ConditionAltKey) @@ -1710,6 +1739,11 @@ func (m *ObjectTransactionResponse) MarshalToSizedBufferVT(dAtA []byte) (int, er i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.FailedMutation != nil { + i = protohelpers.EncodeVarint(dAtA, i, uint64(*m.FailedMutation)) + i-- + dAtA[i] = 0x18 + } if m.ErrorCode != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.ErrorCode)) i-- @@ -7191,6 +7225,14 @@ func (m *ObjectMutation) SizeVT() (n int) { if m.KeepRemoteObject { n += 2 } + if m.Condition != nil { + l = m.Condition.SizeVT() + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } + l = len(m.ConditionKey) + if l > 0 { + n += 2 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -7297,6 +7339,9 @@ func (m *ObjectTransactionRequest) SizeVT() (n int) { if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + if m.ConditionPreferAlt { + n += 2 + } n += len(m.unknownFields) return n } @@ -7314,6 +7359,9 @@ func (m *ObjectTransactionResponse) SizeVT() (n int) { if m.ErrorCode != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.ErrorCode)) } + if m.FailedMutation != nil { + n += 1 + protohelpers.SizeOfVarint(uint64(*m.FailedMutation)) + } n += len(m.unknownFields) return n } @@ -13244,6 +13292,74 @@ func (m *ObjectMutation) UnmarshalVT(dAtA []byte) error { } } m.KeepRemoteObject = bool(v != 0) + case 15: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Condition", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Condition == nil { + m.Condition = &WriteCondition{} + } + if err := m.Condition.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 16: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ConditionKey", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ConditionKey = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) @@ -14097,6 +14213,26 @@ func (m *ObjectTransactionRequest) UnmarshalVT(dAtA []byte) error { } m.ConditionAltKey = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 11: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ConditionPreferAlt", 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.ConditionPreferAlt = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) @@ -14199,6 +14335,26 @@ func (m *ObjectTransactionResponse) UnmarshalVT(dAtA []byte) error { break } } + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field FailedMutation", wireType) + } + var v int32 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.FailedMutation = &v default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 710b9db959..753246698a 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -689,6 +689,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl // retries. With no owner yet (no ring), keep the lock as the bootstrap path. owner := s3a.objectWriteOwner(*input.Bucket, *input.Key) routeKey := s3a.objectRouteKey(*input.Bucket, *input.Key) + // The write condition rides the commit transaction so the owner filer + // evaluates it under the object's path lock — the gateway's earlier + // checkConditionalHeaders is only a fast-path. Conditions it cannot reduce + // (time-based, combined, ETag lists, weak ETags) keep the pre-change + // semantics: check and commit run under the distributed write lock, which + // routed writes still bypass. + cond, reducible := buildWriteCondition(r) completionBody := func() s3err.ErrorCode { var prepCode s3err.ErrorCode completionState, output, prepCode = s3a.prepareMultipartCompletionState(r, input, parts, uploadDirectory, entryName, dirName, completedPartNumbers, completedPartMap, maxPartNo) @@ -796,7 +803,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err) return s3err.ErrInternalError } - if code := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat, versionDir, versionFileName, completionState.finalParts, decorateVersionEntry, *input.UploadId); code != s3err.ErrNone { + if code := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat, versionDir, versionFileName, completionState.finalParts, decorateVersionEntry, *input.UploadId, cond); code != s3err.ErrNone { if code == s3err.ErrNoSuchUpload { return code } @@ -850,6 +857,9 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err) return s3err.ErrInternalError } + // The condition checks the null slot first, then the latest version, + // same resolution the gateway's conditional check performs. + versionsDir := dirName + "/" + entryName + s3_constants.VersionsFolder if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) { if entry.Extended == nil { entry.Extended = make(map[string][]byte) @@ -898,10 +908,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl entry.Attributes.Mime = completionState.mime } entry.Attributes.FileSize = uint64(completionState.offset) - }, removal); err != nil { + }, removal, cond, versionsDir); err != nil { if errors.Is(err, errUploadRemoved) { return s3err.ErrNoSuchUpload } + if errors.Is(err, errObjectPrecondition) { + return s3err.ErrPreconditionFailed + } // The transaction may have committed the object before failing on // the upload removal; a surviving entry references the manifests. if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists { @@ -991,10 +1004,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl if completionState.entityWithTtl { entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true") } - }, removal); err != nil { + }, removal, cond, ""); err != nil { if errors.Is(err, errUploadRemoved) { return s3err.ErrNoSuchUpload } + if errors.Is(err, errObjectPrecondition) { + return s3err.ErrPreconditionFailed + } // The transaction may have committed the object before failing on // the upload removal; a surviving entry references the manifests. if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists { @@ -1017,7 +1033,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl return s3err.ErrNone } var finalizeCode s3err.ErrorCode - if owner != "" { + if owner != "" && reducible { if code := s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key); code != s3err.ErrNone { finalizeCode = code } else { diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 1b716126d9..2990eff785 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -742,7 +742,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion if owner != "" { err = s3a.routedMkFile(owner, s3a.objectRouteKey(dstBucket, normalizedObject), bucketDir+"/"+normalizedObject+s3_constants.VersionsFolder, versionFileName, dstEntry.Chunks, func(entry *filer_pb.Entry) { copyEntryToTarget(entry, dstEntry) - }, nil) + }, nil, nil, "") } else { err = s3a.mkFile(bucketDir, versionObjectPath, dstEntry.Chunks, func(entry *filer_pb.Entry) { copyEntryToTarget(entry, dstEntry) @@ -779,7 +779,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion owner := s3a.objectWriteOwner(dstBucket, normalizedObject) if err = s3a.writeMultipartObject(owner, s3a.objectRouteKey(dstBucket, normalizedObject), dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) { copyEntryToTarget(entry, dstEntry) - }, nil); err != nil { + }, nil, nil, ""); err != nil { return "", "", err } @@ -799,7 +799,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion // default filer is invisible to the next copy until replication. if err = s3a.writeMultipartObject(s3a.objectWriteOwner(dstBucket, normalizedObject), s3a.objectRouteKey(dstBucket, normalizedObject), dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) { copyEntryToTarget(entry, dstEntry) - }, nil); err != nil { + }, nil, nil, ""); err != nil { return "", "", err } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index abe4e98607..502c73e7a5 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -1084,7 +1084,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader lockKey, finalizeMutations = finalize.lockKey, finalize.mutations } routeKey := s3a.objectRouteKey(bucket, object) - resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations) + resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, false, nil, "", finalizeMutations) if err == nil && resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && condAltKey != "" { // A filer that ignores condition_alt_key evaluated the empty // null slot alone; resend the transaction against the entry the @@ -1097,7 +1097,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader conditionKey = condAltKey + "/" + resolved.Name } glog.Warningf("putToFiler: routed PUT to %s precondition-failed for %s; resending against %s", owner, filePath, conditionKey) - resp, err = s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, conditionKey, condAltKey, finalizeMutations) + resp, err = s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, conditionKey, condAltKey, false, nil, "", finalizeMutations) } switch { case err != nil: diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index be70ed1fcb..5c5b5bee7d 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -216,28 +216,37 @@ func (s3a *S3ApiServer) withWriteOwnerFilerClient(owner pb.ServerAddress, fn fun // object path plus a RECOMPUTE_LATEST finalize, so the version's PUT and its // .versions pointer flip commit atomically (the recompute scans .versions/ after // the PUT and sees the new version). -func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey, conditionAltKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) { +// putCondition, when set, guards the PUT mutation itself against putConditionKey +// (e.g. the upload directory a multipart commit must still find); the request +// condition keeps guarding the object the transaction is about. +func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey, conditionAltKey string, conditionPreferAlt bool, putCondition *filer_pb.WriteCondition, putConditionKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) { mutations := make([]*filer_pb.ObjectMutation, 0, 1+len(finalize)) mutations = append(mutations, &filer_pb.ObjectMutation{ - Type: filer_pb.ObjectMutation_PUT, - Directory: path.Dir(filePath), - Entry: entry, + Type: filer_pb.ObjectMutation_PUT, + Directory: path.Dir(filePath), + Entry: entry, + Condition: putCondition, + ConditionKey: putConditionKey, }) mutations = append(mutations, finalize...) return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ - LockKey: lockKey, - RouteKey: routeKey, - Condition: cond, - ConditionKey: conditionKey, - ConditionAltKey: conditionAltKey, - Mutations: mutations, + LockKey: lockKey, + RouteKey: routeKey, + Condition: cond, + ConditionKey: conditionKey, + ConditionAltKey: conditionAltKey, + ConditionPreferAlt: conditionPreferAlt, + Mutations: mutations, }) } // routedMkFile builds an entry like filer_pb.MkFile and writes it through a // routed PUT on the owner filer, for callers that would otherwise mkFile to the -// default filer (e.g. multipart completion of a non-versioned object). -func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn) error { +// default filer (e.g. multipart completion of a non-versioned object). cond is +// the request's write condition, evaluated on the object under the transaction +// lock; condAltKey falls back to the .versions directory when the object slot +// is empty (suspended or versioned buckets). +func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn, cond *filer_pb.WriteCondition, condAltKey string) error { now := time.Now().Unix() entry := &filer_pb.Entry{ Name: name, @@ -254,14 +263,14 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir fn(entry) } filePath := parentDir + "/" + name - var cond *filer_pb.WriteCondition - var conditionKey string + var putCondition *filer_pb.WriteCondition + var putConditionKey string var finalize []*filer_pb.ObjectMutation if removal != nil { - cond, conditionKey = removal.condition, removal.conditionKey + putCondition, putConditionKey = removal.putCondition, removal.putConditionKey finalize = []*filer_pb.ObjectMutation{removal.mutation} } - resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize) + resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, "", condAltKey, false, putCondition, putConditionKey, finalize) if err != nil { return err } @@ -269,12 +278,15 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir // Other keys are nested under this one, so the object goes onto the // directory they live in — the same retry mkFile makes off the ring. entry.MarkPrefixObject() - resp, err = s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize) + resp, err = s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, "", condAltKey, false, putCondition, putConditionKey, finalize) if err != nil { return err } } if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { + if resp.FailedMutation == nil { + return errObjectPrecondition + } return errUploadRemoved } if resp.Error != "" { @@ -283,32 +295,40 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir return nil } -// errUploadRemoved is a routed commit rejected because its precondition found -// the upload directory already deleted, mapped to NoSuchUpload by callers. +// errUploadRemoved is a routed commit rejected because the upload directory's +// precondition found it already deleted, mapped to NoSuchUpload by callers. var errUploadRemoved = errors.New("upload directory already removed") +// errObjectPrecondition is a routed commit rejected because the request's write +// condition (e.g. If-None-Match: *) found the object slot already taken by a +// racing commit, mapped to PreconditionFailed by callers. +var errObjectPrecondition = errors.New("object precondition failed") + // writeMultipartObject writes a completed multipart object entry, routed to the // owner when known (so it serializes with concurrent writes to the same key) // and falling back to a plain mkFile otherwise. removal carries the upload // precondition and deletion applied in the same transaction as the routed PUT. // routeKey must be the same key the caller used to resolve owner, so owner -// selection and forwarding stay consistent. -func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn) error { +// selection and forwarding stay consistent. cond is the request's write +// condition, enforced on the owner under the transaction lock; condAltKey is +// the .versions directory a conditional check falls back to when the bucket +// tracks versions. +func (s3a *S3ApiServer) writeMultipartObject(owner pb.ServerAddress, routeKey, dir, name string, chunks []*filer_pb.FileChunk, fn func(*filer_pb.Entry), removal *uploadRemovalTxn, cond *filer_pb.WriteCondition, condAltKey string) error { if owner != "" { - return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn, removal) + return s3a.routedMkFile(owner, routeKey, dir, name, chunks, fn, removal, cond, condAltKey) } return s3a.mkFile(dir, name, chunks, fn) } -// uploadRemovalTxn carries the precondition and mutation that drop a completed -// upload's directory inside the object's commit transaction. The precondition -// fails the commit when a delete that does not take the object lock (abort, -// lifecycle, s3.clean.uploads) removed the directory first, instead of -// publishing the object over freed chunks. +// uploadRemovalTxn carries the PUT precondition and mutation that drop a +// completed upload's directory inside the object's commit transaction. The +// precondition fails the commit when a delete that does not take the object +// lock (abort, lifecycle, s3.clean.uploads) removed the directory first, +// instead of publishing the object over freed chunks. type uploadRemovalTxn struct { - condition *filer_pb.WriteCondition - conditionKey string - mutation *filer_pb.ObjectMutation + putCondition *filer_pb.WriteCondition + putConditionKey string + mutation *filer_pb.ObjectMutation } // routedUploadRemoval returns the transaction parts that remove a completed @@ -319,14 +339,20 @@ func (s3a *S3ApiServer) routedUploadRemoval(ctx context.Context, owner pb.Server if owner == "" { return nil, nil } + // Older filers ignore the mutation's upload-exists condition, so during an + // upgrade this gateway-side check keeps the same guard window the + // pre-transaction code had. + if exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadId, true); err == nil && !exists { + return nil, errUploadRemoved + } if err := s3a.deleteUnusedPartEntries(ctx, uploadDirectory, bucket, uploadId, completionState); err != nil { return nil, err } conditionKey, condition := uploadExistsCondition(uploadDirectory) return &uploadRemovalTxn{ - condition: condition, - conditionKey: conditionKey, - mutation: s3a.removeUploadDirMutation(bucket, uploadId), + putCondition: condition, + putConditionKey: conditionKey, + mutation: s3a.removeUploadDirMutation(bucket, uploadId), }, nil } diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 0f23779d72..19f202c865 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -514,7 +514,7 @@ func TestRoutedMultipartFinalize(t *testing.T) { chunks := []*filer_pb.FileChunk{{FileId: "1,01637037d6"}} code := s3a.routedMultipartFinalize(owner, "b", "obj", false, "/buckets/b/obj.versions", "v_1", chunks, func(entry *filer_pb.Entry) { entry.Extended = map[string][]byte{s3_constants.SeaweedFSUploadId: []byte("up1")} - }, "up1") + }, "up1", clause(filer_pb.WriteCondition_IF_NOT_EXISTS)) if code != s3err.ErrNone { t.Fatalf("routedMultipartFinalize = %v", code) } @@ -529,10 +529,15 @@ func TestRoutedMultipartFinalize(t *testing.T) { if req.RouteKey != "s3.object.write:/buckets/b/obj" { t.Fatalf("RouteKey = %q", req.RouteKey) } - if req.ConditionKey != "/buckets/b/.uploads/up1" { + // The request condition guards the object (at lock_key, falling back to + // .versions); the upload directory rides the PUT as a mutation guard. + if req.ConditionKey != "" { t.Fatalf("ConditionKey = %q", req.ConditionKey) } - if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + if req.ConditionAltKey != "/buckets/b/obj.versions" { + t.Fatalf("ConditionAltKey = %q", req.ConditionAltKey) + } + if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_NOT_EXISTS { t.Fatalf("Condition = %+v", req.Condition) } if len(req.Mutations) != 3 { @@ -548,6 +553,10 @@ func TestRoutedMultipartFinalize(t *testing.T) { string(put.Entry.Extended[s3_constants.SeaweedFSUploadId]) != "up1" { t.Fatalf("put mutation = %+v", put) } + if put.ConditionKey != "/buckets/b/.uploads/up1" || + put.Condition == nil || len(put.Condition.Clauses) != 1 || put.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("put condition = %+v key %q", put.Condition, put.ConditionKey) + } removeUpload := req.Mutations[1] if removeUpload.Type != filer_pb.ObjectMutation_DELETE || removeUpload.Directory != "/buckets/b/.uploads" || @@ -580,7 +589,7 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) { if err != nil { t.Fatalf("routedUploadRemoval: %v", err) } - if err := s3a.writeMultipartObject(owner, "s3.object.write:/buckets/b/o", "/buckets/b", "o", nil, nil, removal); err != nil { + if err := s3a.writeMultipartObject(owner, "s3.object.write:/buckets/b/o", "/buckets/b", "o", nil, nil, removal, nil, ""); err != nil { t.Fatalf("writeMultipartObject: %v", err) } @@ -591,12 +600,6 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) { if req.LockKey != "/buckets/b/o" { t.Fatalf("LockKey = %q", req.LockKey) } - if req.ConditionKey != "/buckets/b/.uploads/up1" { - t.Fatalf("ConditionKey = %q", req.ConditionKey) - } - if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { - t.Fatalf("Condition = %+v", req.Condition) - } if len(req.Mutations) != 2 { t.Fatalf("mutations = %d, want 2", len(req.Mutations)) } @@ -604,6 +607,10 @@ func TestWriteMultipartObjectRemovesUploadDir(t *testing.T) { if put.Type != filer_pb.ObjectMutation_PUT || put.Directory != "/buckets/b" || put.Entry == nil || put.Entry.Name != "o" { t.Fatalf("put mutation = %+v", put) } + if put.ConditionKey != "/buckets/b/.uploads/up1" || + put.Condition == nil || len(put.Condition.Clauses) != 1 || put.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS { + t.Fatalf("put condition = %+v key %q", put.Condition, put.ConditionKey) + } removeUpload := req.Mutations[1] if removeUpload.Type != filer_pb.ObjectMutation_DELETE || removeUpload.Directory != "/buckets/b/.uploads" || diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index 4f470a71de..c9ec86251a 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -116,8 +116,10 @@ func uploadExistsCondition(uploadDirectory string) (string, *filer_pb.WriteCondi // reachable: PUT first keeps the chunks referenced at all times, and removing // the upload directory before the recompute means a published object never // coexists with a leftover .uploads directory that s3.clean.uploads would purge -// with data. -func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool, versionDir, versionFileName string, chunks []*filer_pb.FileChunk, decorate func(*filer_pb.Entry), uploadID string) s3err.ErrorCode { +// with data. cond is the request's write condition, checked against the object +// (its latest version when the slot at the object path is empty) under the same +// lock; the upload directory rides the PUT as a mutation precondition. +func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool, versionDir, versionFileName string, chunks []*filer_pb.FileChunk, decorate func(*filer_pb.Entry), uploadID string, cond *filer_pb.WriteCondition) s3err.ErrorCode { now := time.Now().Unix() versionEntry := &filer_pb.Entry{ Name: versionFileName, @@ -138,8 +140,14 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, if owner != "" { routeKey = s3a.objectRouteKey(bucket, object) } - conditionKey, condition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID) - resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, "", []*filer_pb.ObjectMutation{ + // A filer older than the mutation-condition support silently ignores + // putCondition, so on upgrade the guard degrades to this gateway-side + // check — the same window the pre-transaction code had. + if exists, err := s3a.exists(s3a.genUploadsFolder(bucket), uploadID, true); err == nil && !exists { + return s3err.ErrNoSuchUpload + } + putConditionKey, putCondition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID) + resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, cond, "", versionDir, true, putCondition, putConditionKey, []*filer_pb.ObjectMutation{ s3a.removeUploadDirMutation(bucket, uploadID), s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true), }) @@ -148,6 +156,9 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, glog.Errorf("routedMultipartFinalize: %s/%s upload %s on %s: %v", bucket, object, uploadID, owner, err) return s3err.ErrInternalError case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED: + if resp.FailedMutation == nil { + return s3err.ErrPreconditionFailed + } return s3err.ErrNoSuchUpload case resp.Error != "": glog.Errorf("routedMultipartFinalize: %s/%s upload %s: %s", bucket, object, uploadID, resp.Error) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 21f823410b..498bdd624c 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -353,6 +353,7 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje IsMoved: true, RequestId: req.RequestId, ConditionAltKey: req.ConditionAltKey, + ConditionPreferAlt: req.ConditionPreferAlt, } var resp *filer_pb.ObjectTransactionResponse handled, err := fs.forwardToWriteOwner(ctx, req.RouteKey, func(owner pb.ServerAddress) error { @@ -407,11 +408,16 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje } current, findErr := fs.filer.FindEntry(ctx, conditionPath) if findErr == nil && current != nil && current.IsDirectory() && req.ConditionAltKey != "" { - // A directory is not the object: implicit directories can share a - // path with an S3 key, so treat it as an absent slot. Only the - // alt-key callers want object-slot semantics; an explicit - // condition_key may legitimately name a directory. - current, findErr = nil, filer_pb.ErrNotFound + // A bare directory is not the object: implicit directories can + // share a path with an S3 key, so treat it as an absent slot. A + // directory entry holding content or the prefix-object marker is + // an object and stays. Only the alt-key callers want object-slot + // semantics; an explicit condition_key may legitimately name a + // directory. + _, isPrefixObject := current.Extended[s3_constants.SeaweedFSPrefixObject] + if current.Mime == "" && len(current.GetChunks()) == 0 && len(current.Content) == 0 && !current.IsInRemoteOnly() && !isPrefixObject { + current, findErr = nil, filer_pb.ErrNotFound + } } // The caller can name a fallback evaluated when the primary is absent — // a suspended-versioning write checks "the null slot if present, else @@ -419,8 +425,24 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje // the object's .versions directory, and the latest it resolves to is // read from the pointer maintained by the same lock, so a concurrent // version delete cannot move which entry the condition sees. - if findErr == filer_pb.ErrNotFound && req.ConditionAltKey != "" { - current, conditionPath, findErr = fs.latestVersionedConditionEntry(ctx, util.FullPath(req.ConditionAltKey)) + if req.ConditionAltKey != "" { + altEntry, altPath, altErr := fs.latestVersionedConditionEntry(ctx, util.FullPath(req.ConditionAltKey)) + if req.ConditionPreferAlt { + // The versioned path resolves "current" through the latest + // pointer, so a stale null-version entry or a tombstoned + // latest decides correctly. When the alt path cannot resolve + // (no .versions at all), the primary entry stands. + switch { + case altErr == nil: + current, conditionPath, findErr = altEntry, altPath, nil + case errors.Is(altErr, filer_pb.ErrNotFound): + default: + record = false + return nil, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, altErr) + } + } else if findErr == filer_pb.ErrNotFound { + current, conditionPath, findErr = altEntry, altPath, altErr + } } if findErr != nil && findErr != filer_pb.ErrNotFound { record = false // a lookup error decides nothing; a retry must re-run, not replay @@ -438,10 +460,26 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje } for i, m := range req.Mutations { + if code, err := fs.checkMutationCondition(ctx, m); err != nil { + glog.V(2).InfofCtx(ctx, "ObjectTransaction %s mutation %d condition lookup: %v", lockPath, i, err) + resp.Error = fmt.Sprintf("mutation %d: %v", i, err) + failed := int32(i) + resp.FailedMutation = &failed + return resp, nil + } else if code != filer_pb.FilerError_OK { + glog.V(3).InfofCtx(ctx, "ObjectTransaction %s mutation %d precondition failed", lockPath, i) + resp.Error = fmt.Sprintf("mutation %d: precondition failed", i) + resp.ErrorCode = code + failed := int32(i) + resp.FailedMutation = &failed + return resp, nil + } if err := fs.applyObjectMutation(ctx, m, req.IsFromOtherCluster, req.Signatures); err != nil { glog.V(2).InfofCtx(ctx, "ObjectTransaction %s mutation %d (%v): %v", lockPath, i, m.Type, err) resp.Error = fmt.Sprintf("mutation %d: %v", i, err) resp.ErrorCode = filerErrorCode(err) + failed := int32(i) + resp.FailedMutation = &failed return resp, nil } } @@ -449,6 +487,36 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje return resp, nil } +// checkMutationCondition evaluates a mutation's optional precondition under the +// transaction's lock, against condition_key when set or the mutation's target +// (the PUT's entry path, else directory/name). Unlike the request-level +// condition there is no alt-key fallback: a mutation guard names the exact entry +// it guards. +func (fs *FilerServer) checkMutationCondition(ctx context.Context, m *filer_pb.ObjectMutation) (filer_pb.FilerError, error) { + if !conditionIsSet(m.Condition) { + return filer_pb.FilerError_OK, nil + } + target := util.FullPath(m.ConditionKey) + if target == "" { + name := m.Name + if m.Type == filer_pb.ObjectMutation_PUT && m.Entry != nil { + name = m.Entry.Name + } + target = util.JoinPath(m.Directory, name) + } + current, err := fs.filer.FindEntry(ctx, target) + if err != nil && err != filer_pb.ErrNotFound { + return filer_pb.FilerError_OK, fmt.Errorf("condition check %s: %w", target, err) + } + if err == filer_pb.ErrNotFound { + current = nil + } + if !writeConditionSatisfied(m.Condition, current) { + return filer_pb.FilerError_PRECONDITION_FAILED, nil + } + return filer_pb.FilerError_OK, nil +} + const ( recentObjectTxnTTL = 2 * time.Minute recentObjectTxnSweepCap = 10000 @@ -525,9 +593,11 @@ func (fs *FilerServer) rememberObjectTxn(requestID string, lockPath util.FullPat // latestVersionedConditionEntry resolves the object a conditional write should // be checked against inside a .versions directory: the entry the latest-version // pointer names, or nothing when the pointer names a delete marker — a -// tombstone is not a current version. A missing pointer or a pointer to a file -// that is gone means the metadata cannot answer the question, which reports an -// error rather than guessing at either branch. +// tombstone is not a current version. A suspended-versioning write clears the +// pointer and stamps null-version-is-latest instead, which reports not-found so +// the regular path's null object decides. Any other missing pointer, or a +// pointer to a file that is gone, means the metadata cannot answer the +// question, which reports an error rather than guessing at either branch. func (fs *FilerServer) latestVersionedConditionEntry(ctx context.Context, versionsDir util.FullPath) (*filer.Entry, util.FullPath, error) { versionsEntry, err := fs.filer.FindEntry(ctx, versionsDir) if err != nil { @@ -538,6 +608,9 @@ func (fs *FilerServer) latestVersionedConditionEntry(ctx context.Context, versio } latestName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]) if latestName == "" { + if string(versionsEntry.Extended[s3_constants.ExtNullVersionIsLatestKey]) == "true" { + return nil, versionsDir, filer_pb.ErrNotFound + } return nil, versionsDir, fmt.Errorf("%s has no latest-version pointer", versionsDir) } latestPath := util.NewFullPath(string(versionsDir), latestName) @@ -549,7 +622,7 @@ func (fs *FilerServer) latestVersionedConditionEntry(ctx context.Context, versio return nil, latestPath, err } if string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { - return nil, latestPath, filer_pb.ErrNotFound + return nil, latestPath, nil } return entry, latestPath, nil } diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index c73badd7a3..b7fd0b5117 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -229,6 +229,90 @@ func TestObjectTransactionPreconditionAborts(t *testing.T) { } } +// A mutation's own precondition is checked under the same lock just before the +// mutation applies; a failure aborts the transaction at that mutation and is +// reported through failed_mutation, leaving earlier mutations' work visible. +func TestObjectTransactionMutationCondition(t *testing.T) { + now := time.Unix(1700000000, 0) + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ + "/buckets/bkt/obj": { + Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}, + }, + }) + + mkReq := func() *filer_pb.ObjectTransactionRequest { + return &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + Mutations: []*filer_pb.ObjectMutation{ + {Type: filer_pb.ObjectMutation_PUT, Directory: "/buckets/bkt", Entry: &filer_pb.Entry{ + Name: "other", + Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 2}, + }, Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{ + {Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}, + }}, ConditionKey: "/buckets/bkt/obj"}, + {Type: filer_pb.ObjectMutation_DELETE, Directory: "/buckets/bkt", Name: "obj"}, + }, + } + } + + resp, err := fs.ObjectTransaction(context.Background(), mkReq()) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("want PRECONDITION_FAILED, got %v (%q)", resp.ErrorCode, resp.Error) + } + if resp.FailedMutation == nil || *resp.FailedMutation != 0 { + t.Fatalf("want failed_mutation=0, got %v", resp.FailedMutation) + } + if _, ok := store.entries["/buckets/bkt/other"]; ok { + t.Errorf("PUT whose condition failed must not store an entry") + } + if _, ok := store.entries["/buckets/bkt/obj"]; !ok { + t.Errorf("a failed mutation condition must abort before later mutations run") + } + + // With the slot absent the same transaction applies both mutations. + delete(store.entries, "/buckets/bkt/obj") + resp, err = fs.ObjectTransaction(context.Background(), mkReq()) + if err != nil || resp.Error != "" { + t.Fatalf("txn failed: err=%v resp=%q", err, resp.Error) + } + if _, ok := store.entries["/buckets/bkt/other"]; !ok { + t.Errorf("condition satisfied: PUT entry missing") + } +} + +// The request-level condition keeps reporting no failed_mutation, so callers +// can tell it apart from a mutation's own guard. +func TestObjectTransactionMutationConditionVsRequest(t *testing.T) { + now := time.Unix(1700000000, 0) + fs, _ := newTxnTestServer(t, map[string]*filer.Entry{ + "/buckets/bkt/obj": { + Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}, + }, + }) + + resp, err := fs.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{ + {Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}, + }}, + Mutations: []*filer_pb.ObjectMutation{ + {Type: filer_pb.ObjectMutation_DELETE, Directory: "/buckets/bkt", Name: "obj"}, + }, + }) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("want PRECONDITION_FAILED, got %v (%q)", resp.ErrorCode, resp.Error) + } + if resp.FailedMutation != nil { + t.Fatalf("request-level failure must not set failed_mutation, got %d", *resp.FailedMutation) + } +} + // Deleting the latest version and recomputing re-points the pointer at the new // highest-named remaining version; the scan runs under the transaction lock. func TestObjectTransactionRecomputeLatest(t *testing.T) { @@ -1095,3 +1179,42 @@ func TestObjectTransactionConditionIgnoresDirectory(t *testing.T) { t.Fatalf("directory at key with live latest: err=%v code=%v, want precondition failed", err, resp.GetErrorCode()) } } + +// A suspended-versioning write clears the latest-version pointer and leaves +// the null object at the regular path as current. The prefer-alt condition +// must keep resolving against that object, not reject on the empty pointer. +func TestObjectTransactionConditionPreferAltHonorsNullIsLatest(t *testing.T) { + now := time.Unix(1700000000, 0) + seed := map[string]*filer.Entry{ + "/buckets/bkt/obj": { + Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0644}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("nulletag")}, + }, + "/buckets/bkt/obj/.versions": { + Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755}, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + }, + } + req := &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + ConditionAltKey: "/buckets/bkt/obj/.versions", + ConditionPreferAlt: true, + Condition: &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: []string{`"nulletag"`}}}, + }, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: "/buckets/bkt", + Entry: &filer_pb.Entry{Name: "obj", Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 9}}, + }}, + } + + fs, store := newTxnTestServer(t, seed) + resp, err := fs.ObjectTransaction(context.Background(), req) + if err != nil || resp.Error != "" { + t.Fatalf("null-is-latest: err=%v resp=%q, want applied", err, resp.Error) + } + if store.entries["/buckets/bkt/obj"] == nil { + t.Fatal("mutation did not apply over a null-is-latest .versions") + } +}