diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index a97fbfdf9..bfa1de291 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -392,6 +392,7 @@ message ObjectTransactionRequest { string condition_key = 6; // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key) string route_key = 7; // ring key identifying the owner filer; a non-owner forwards the whole transaction to it bool is_moved = 8; // set on a forwarded transaction so the receiver applies it locally instead of forwarding again + 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) } message ObjectTransactionResponse { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index fe0d0767a..1d82746e2 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1811,9 +1811,10 @@ 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 + 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 + 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) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1904,6 +1905,13 @@ func (x *ObjectTransactionRequest) GetIsMoved() bool { return false } +func (x *ObjectTransactionRequest) GetConditionAltKey() string { + if x != nil { + return x.ConditionAltKey + } + return "" +} + type ObjectTransactionResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` @@ -7433,7 +7441,7 @@ const file_filer_proto_rawDesc = "" + "\fexclude_name\x18\t \x01(\tR\vexcludeName\x1a?\n" + "\x11CopyExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xd5\x02\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x81\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" + @@ -7444,7 +7452,9 @@ const file_filer_proto_rawDesc = "" + "signatures\x12#\n" + "\rcondition_key\x18\x06 \x01(\tR\fconditionKey\x12\x1b\n" + "\troute_key\x18\a \x01(\tR\brouteKey\x12\x19\n" + - "\bis_moved\x18\b \x01(\bR\aisMoved\"f\n" + + "\bis_moved\x18\b \x01(\bR\aisMoved\x12*\n" + + "\x11condition_alt_key\x18\n" + + " \x01(\tR\x0fconditionAltKey\"f\n" + "\x19ObjectTransactionResponse\x12\x14\n" + "\x05error\x18\x01 \x01(\tR\x05error\x123\n" + "\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 46d4b6e0c..e11dd0045 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1567,6 +1567,13 @@ func (m *ObjectTransactionRequest) MarshalToSizedBufferVT(dAtA []byte) (int, err i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.ConditionAltKey) > 0 { + i -= len(m.ConditionAltKey) + copy(dAtA[i:], m.ConditionAltKey) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.ConditionAltKey))) + i-- + dAtA[i] = 0x52 + } if m.IsMoved { i-- if m.IsMoved { @@ -7188,6 +7195,10 @@ func (m *ObjectTransactionRequest) SizeVT() (n int) { if m.IsMoved { n += 2 } + l = len(m.ConditionAltKey) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -13841,6 +13852,38 @@ func (m *ObjectTransactionRequest) UnmarshalVT(dAtA []byte) error { } } m.IsMoved = bool(v != 0) + case 10: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ConditionAltKey", 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.ConditionAltKey = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 3cb1caba5..9a6f900a6 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -478,6 +478,10 @@ type putFinalize struct { lockKey string mutations []*filer_pb.ObjectMutation afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode + // conditionAltKey names the entry a routed WriteCondition falls back to + // when the write path is absent — a suspended-versioning write points it + // at the latest version so check and write stay under the filer lock + conditionAltKey string } const ( @@ -1010,15 +1014,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader routed := false routedAmbiguous := false var owner pb.ServerAddress + condAltKey := "" + if finalize != nil { + condAltKey = finalize.conditionAltKey + } if owner = s3a.routableWriteOwner(bucket, object); owner != "" { - if cond, ok := routeWriteCondition(r, uniqueWritePath); ok { + if cond, ok := routeWriteCondition(r, uniqueWritePath && condAltKey == ""); ok { // Routed mutations ride in the PUT's transaction (committing atomically), // so lockKey is the object path they carry, not the version file path. lockKey, finalizeMutations := filePath, []*filer_pb.ObjectMutation(nil) if finalize != nil && len(finalize.mutations) > 0 { lockKey, finalizeMutations = finalize.lockKey, finalize.mutations } - resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", finalizeMutations) + resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations) switch { case err != nil: routedAmbiguous = true @@ -1824,7 +1832,19 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob // naming a marker that is gone, and the read path heals that by promoting an older // version, republishing the deleted key. Best-effort, as a stale flag self-heals on // the next list. + // A conditional suspended write checks "the null slot if present, else the + // latest version". Naming the pointed-at version as the fallback lets the + // routed transaction evaluate that rule under the filer lock instead of + // splitting check and write across a gateway lock that routed writes skip. + conditionAltKey := "" + if headers, headersErr := parseConditionalHeaders(r); headersErr == s3err.ErrNone && headers.isSet { + // The filer resolves the latest pointer under the same lock that + // serializes the write, so the condition cannot read a version a + // concurrent delete has already displaced. + conditionAltKey = s3a.toFilerPath(bucket, normalizedObject+s3_constants.VersionsFolder) + } etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, &putFinalize{ + conditionAltKey: conditionAltKey, afterCreate: func(entry *filer_pb.Entry) s3err.ErrorCode { writtenETag := string(entry.Extended[s3_constants.ExtETagKey]) if err := s3a.finalizeSuspendedNullWrite("", bucket, normalizedObject, s3_constants.ExtETagKey, writtenETag); err != nil { @@ -1832,7 +1852,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob } return s3err.ErrNone }, - }, false, "") + }, true, "") if errCode != s3err.ErrNone { glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode) return "", errCode, SSEResponseMetadata{} diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 07b131f44..b2c236f9a 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -8,9 +8,12 @@ import ( "net" "net/http" "net/http/httptest" + "path" + "slices" "strings" "sync" "testing" + "time" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -18,10 +21,12 @@ import ( "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/wdclient" @@ -112,6 +117,8 @@ type ambiguousPutFiler struct { mu sync.Mutex entries map[string]*filer_pb.Entry + txnCount int + txnCheck func(req *filer_pb.ObjectTransactionRequest) apply bool createErr error respError string @@ -163,6 +170,92 @@ func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_p return &filer_pb.LookupDirectoryEntryResponse{}, nil } +// ObjectTransaction applies the WriteCondition against the entry the condition +// key names, the way a real filer evaluates it under the path lock, then +// applies PUT mutations. Routed-write tests use it to catch a condition sent +// for an entry the gateway's conditional check never looked at. +func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.txnCount++ + if f.txnCheck != nil { + f.txnCheck(req) + } + key := req.LockKey + if req.ConditionKey != "" { + key = req.ConditionKey + } + current := f.entries[key] + if current == nil && req.ConditionAltKey != "" { + // Mirrors the filer's fallback: the alt key is the object's .versions + // directory and the entry under test is the one its latest pointer + // names; a delete marker resolves to absent. + if versionsEntry := f.entries[req.ConditionAltKey]; versionsEntry != nil { + if latest := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]); latest != "" { + key = req.ConditionAltKey + "/" + latest + current = f.entries[key] + if current != nil && string(current.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + current = nil + } + } + } + } + for _, clause := range req.GetCondition().GetClauses() { + var md5Hex string + if current != nil && current.Attributes != nil { + md5Hex = fmt.Sprintf("%x", current.Attributes.Md5) + } + ok := true + switch clause.Kind { + case filer_pb.WriteCondition_IF_NOT_EXISTS: + ok = current == nil + case filer_pb.WriteCondition_IF_EXISTS: + ok = current != nil + case filer_pb.WriteCondition_IF_ETAG_MATCH: + ok = current != nil && slices.Contains(clause.Etags, md5Hex) + case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH: + ok = current == nil || !slices.Contains(clause.Etags, md5Hex) + } + if !ok { + return &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } + } + for _, m := range req.Mutations { + if m.Type == filer_pb.ObjectMutation_PUT { + f.entries[m.Directory+"/"+m.Entry.Name] = proto.Clone(m.Entry).(*filer_pb.Entry) + } + } + return &filer_pb.ObjectTransactionResponse{}, nil +} + +func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { + f.mu.Lock() + defer f.mu.Unlock() + for fullPath, entry := range f.entries { + if path.Dir(fullPath) != req.Directory { + continue + } + if req.Prefix != "" && !strings.HasPrefix(entry.Name, req.Prefix) { + continue + } + if req.StartFromFileName != "" && (entry.Name < req.StartFromFileName || + (entry.Name == req.StartFromFileName && !req.InclusiveStartFrom)) { + continue + } + if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}); err != nil { + return err + } + } + return nil +} + +func (f *ambiguousPutFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.entries[req.Directory+"/"+req.Entry.Name] = proto.Clone(req.Entry).(*filer_pb.Entry) + return &filer_pb.UpdateEntryResponse{}, nil +} + func (f *ambiguousPutFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { resp := &filer_pb.LookupVolumeResponse{LocationsMap: map[string]*filer_pb.Locations{}} for _, vid := range req.VolumeIds { @@ -407,3 +500,115 @@ func TestFetchChunkDataDecryptsVolumeCipher(t *testing.T) { t.Fatalf("ranged plain read = %q, want %q", got, plaintext[5:9]) } } + +// A suspended-versioning overwrite resolves the conditional check against the +// entry the read path serves: the null slot when present, else the latest +// version under .versions/. The routed transaction carries that version as the +// condition's fallback key so check and write stay under the filer lock — +// checking the empty null-version slot alone would reject a matching If-Match. +func TestSuspendedPutIfMatchRoutesWithConditionAltKey(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningSuspended)}, + }, + "/buckets/b/k.versions": { + Name: "k.versions", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("v1"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"), + }, + }, + "/buckets/b/k.versions/v_v1": { + Name: "v_v1", + Attributes: &filer_pb.FuseAttributes{Md5: []byte{0xab, 0xc1, 0x23}, FileSize: 1}, + Extended: map[string][]byte{ + s3_constants.ExtETagKey: []byte("abc123"), + s3_constants.ExtVersionIdKey: []byte("v1"), + }, + }, + }, + apply: true, + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { + if req.ConditionAltKey != "/buckets/b/k.versions" { + t.Fatalf("condition alt key = %q, want /buckets/b/k.versions", req.ConditionAltKey) + } + } + + r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new")) + r.Header.Set(s3_constants.IfMatch, `"abc123"`) + _, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "") + if code != s3err.ErrNone { + t.Fatalf("suspended If-Match PUT = %v", code) + } + if filerImpl.entries["/buckets/b/k"] == nil { + t.Fatal("null-version entry was not created") + } + if filerImpl.txnCount != 1 { + t.Fatalf("conditional suspended write should route one transaction, got %d", filerImpl.txnCount) + } +} + +// When the latest version is a delete marker, a suspended If-None-Match: * +// write must treat the object as absent — the fallback resolves a tombstone +// to "no current version", not an existing one. +func TestSuspendedPutIfNoneMatchWithDeleteMarkerLatest(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningSuspended)}, + }, + "/buckets/b/k.versions": { + Name: "k.versions", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"), + }, + }, + "/buckets/b/k.versions/v_v1": { + Name: "v_v1", + Attributes: &filer_pb.FuseAttributes{FileSize: 0}, + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("v1"), + s3_constants.ExtDeleteMarkerKey: []byte("true"), + }, + }, + }, + apply: true, + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new")) + r.Header.Set(s3_constants.IfNoneMatch, "*") + _, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "") + if code != s3err.ErrNone { + t.Fatalf("suspended If-None-Match PUT over a delete-marker latest = %v, want success", code) + } + if filerImpl.entries["/buckets/b/k"] == nil { + t.Fatal("null-version entry was not created") + } +} diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 67c027908..7a323f1f5 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -183,7 +183,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O // 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 string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) { +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) { mutations := make([]*filer_pb.ObjectMutation, 0, 1+len(finalize)) mutations = append(mutations, &filer_pb.ObjectMutation{ Type: filer_pb.ObjectMutation_PUT, @@ -192,11 +192,12 @@ func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, fil }) mutations = append(mutations, finalize...) return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ - LockKey: lockKey, - RouteKey: routeKey, - Condition: cond, - ConditionKey: conditionKey, - Mutations: mutations, + LockKey: lockKey, + RouteKey: routeKey, + Condition: cond, + ConditionKey: conditionKey, + ConditionAltKey: conditionAltKey, + Mutations: mutations, }) } @@ -227,7 +228,7 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir cond, conditionKey = removal.condition, removal.conditionKey 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, conditionKey, "", finalize) if err != nil { return err } diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index 5f4726331..e37d5d6f8 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -139,7 +139,7 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket, 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{ + resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, "", []*filer_pb.ObjectMutation{ s3a.removeUploadDirMutation(bucket, uploadID), s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true), }) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 862b9b4fa..a5dfdd7fc 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -18,6 +18,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/wdclient" @@ -343,6 +344,7 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje ConditionKey: req.ConditionKey, RouteKey: req.RouteKey, IsMoved: true, + ConditionAltKey: req.ConditionAltKey, } var resp *filer_pb.ObjectTransactionResponse handled, err := fs.forwardToWriteOwner(ctx, req.RouteKey, func(owner pb.ServerAddress) error { @@ -374,6 +376,22 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje conditionPath = util.FullPath(req.ConditionKey) } 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 + } + // The caller can name a fallback evaluated when the primary is absent — + // a suspended-versioning write checks "the null slot if present, else + // the latest version" atomically under this lock. condition_alt_key is + // 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 findErr != nil && findErr != filer_pb.ErrNotFound { return &filer_pb.ObjectTransactionResponse{}, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, findErr) } @@ -399,6 +417,38 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje return &filer_pb.ObjectTransactionResponse{}, nil } +// 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. +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 { + return nil, versionsDir, err + } + if versionsEntry == nil { + return nil, versionsDir, filer_pb.ErrNotFound + } + latestName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]) + if latestName == "" { + return nil, versionsDir, fmt.Errorf("%s has no latest-version pointer", versionsDir) + } + latestPath := util.NewFullPath(string(versionsDir), latestName) + entry, err := fs.filer.FindEntry(ctx, latestPath) + if err == filer_pb.ErrNotFound { + return nil, latestPath, fmt.Errorf("latest-version pointer %s names a missing entry", latestPath) + } + if err != nil { + return nil, latestPath, err + } + if string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + return nil, latestPath, filer_pb.ErrNotFound + } + return entry, latestPath, nil +} + // ObjectTransactionBatch applies several object transactions in one round trip, // each under its own per-path lock and independent of the others. A failed // transaction (precondition or mutation error) is reported in its own response diff --git a/weed/server/filer_grpc_server_object_txn_test.go b/weed/server/filer_grpc_server_object_txn_test.go index ad2c56129..51b1442fb 100644 --- a/weed/server/filer_grpc_server_object_txn_test.go +++ b/weed/server/filer_grpc_server_object_txn_test.go @@ -3,7 +3,9 @@ package weed_server import ( "context" "net" + "os" "strconv" + "strings" "testing" "time" @@ -852,3 +854,110 @@ func TestObjectTransactionForwardsToOwner(t *testing.T) { t.Errorf("non-owner must forward, not apply locally") } } + +// A suspended-versioning conditional write falls back to the latest version +// under the transaction lock: a real latest satisfies existence checks, and a +// delete marker reads as absent. +func TestObjectTransactionConditionFallbackResolvesLatestPointer(t *testing.T) { + now := time.Unix(1700000000, 0) + seed := func(latestName, marker string) map[string]*filer.Entry { + entries := map[string]*filer.Entry{ + "/buckets/bkt/obj/.versions": { + Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755}, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte(latestName)}, + }, + } + ext := map[string][]byte{s3_constants.ExtETagKey: []byte("abc")} + if marker != "" { + ext[s3_constants.ExtDeleteMarkerKey] = []byte(marker) + } + entries["/buckets/bkt/obj/.versions/"+latestName] = &filer.Entry{ + Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644}, + Extended: ext, + } + return entries + } + req := func() *filer_pb.ObjectTransactionRequest { + return &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + ConditionAltKey: "/buckets/bkt/obj/.versions", + Condition: &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}}, + }, + 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("v_live", "")) + resp, err := fs.ObjectTransaction(context.Background(), req()) + if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("latest version present: err=%v code=%v, want precondition failed", err, resp.GetErrorCode()) + } + + fs, store = newTxnTestServer(t, seed("v_gone", "true")) + resp, err = fs.ObjectTransaction(context.Background(), req()) + if err != nil || resp.Error != "" { + t.Fatalf("delete-marker 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 delete-marker latest") + } +} + +// A bare directory at the object path — left over when /bkt/obj/child was +// created — is not the object, so the condition must fall back to the +// .versions resolution instead of counting the directory as an existing key. +func TestObjectTransactionConditionIgnoresDirectory(t *testing.T) { + now := time.Unix(1700000000, 0) + seed := func(marker string) map[string]*filer.Entry { + return map[string]*filer.Entry{ + "/buckets/bkt/obj": { + Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0755 | os.ModeDir}, + }, + "/buckets/bkt/obj/.versions": { + Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755}, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v1")}, + }, + "/buckets/bkt/obj/.versions/v1": { + Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc"), s3_constants.ExtDeleteMarkerKey: []byte(marker)}, + }, + } + } + req := &filer_pb.ObjectTransactionRequest{ + LockKey: "/buckets/bkt/obj", + ConditionAltKey: "/buckets/bkt/obj/.versions", + Condition: &filer_pb.WriteCondition{ + Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}}, + }, + 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, _ := newTxnTestServer(t, seed("true")) + resp, err := fs.ObjectTransaction(context.Background(), req) + if err != nil { + t.Fatalf("directory at key with delete-marker latest: %v", err) + } + // The condition must pass so the write reaches the type-conflict path; + // the directory itself rejects the PUT, which is the caller's business. + if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("condition must not fail on a bare directory: resp=%q", resp.Error) + } + if !strings.Contains(resp.Error, "is a directory") { + t.Fatalf("want the dir/file type-conflict error, got %q", resp.Error) + } + + fs, _ = newTxnTestServer(t, seed("")) + resp, err = fs.ObjectTransaction(context.Background(), req) + if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED { + t.Fatalf("directory at key with live latest: err=%v code=%v, want precondition failed", err, resp.GetErrorCode()) + } +}