diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index b9fdf46dd1..0021c2ec52 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -379,6 +379,7 @@ message Recompute { string demote_key = 7; // if set, stamp demote_value on the prior name_to_key target when it changes bytes demote_value = 8; // value for demote_key string exclude_name = 9; // if set, skip this child when scanning (e.g. a version about to be deleted) + string only_if_current_name = 10; // if set, apply only while the pointer's name_to_key value still equals this } // ObjectTransactionRequest applies an ordered list of mutations atomically with diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 0330ec2395..06168a8006 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -522,6 +522,25 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, glog.V(3).InfofCtx(ctx, "EEXIST: entry %s already exists", entry.FullPath) return fmt.Errorf("%s: %w", entry.FullPath, filer_pb.ErrEntryAlreadyExists) } + // An object landing on a directory (prefix object) replaces the object + // data only: the directory keeps its own mode, owner and xattrs. Only + // xattr- keys merge — FUSE stores them on the directory itself — while a + // marker's other fields (encryption, version, etag) belong to the + // object it held, not to the directory. + if oldEntry.IsDirectory() && entry.IsDirectory() { + if _, marked := entry.Extended[s3_constants.SeaweedFSPrefixObject]; marked { + entry.Attr.Mode = oldEntry.Attr.Mode + entry.Attr.Uid = oldEntry.Attr.Uid + entry.Attr.Gid = oldEntry.Attr.Gid + entry.Attr.UserName = oldEntry.Attr.UserName + entry.Attr.GroupNames = oldEntry.Attr.GroupNames + for key, value := range oldEntry.Extended { + if _, ok := entry.Extended[key]; !ok && strings.HasPrefix(key, "xattr-") { + entry.Extended[key] = value + } + } + } + } glog.V(4).InfofCtx(ctx, "UpdateEntry %s: old entry: %v", entry.FullPath, oldEntry.Name()) f.stampEntryCommitTsNs(entry) stamped = true diff --git a/weed/filer/filer_prefix_object_test.go b/weed/filer/filer_prefix_object_test.go index 3d399660f4..fbfb1ee601 100644 --- a/weed/filer/filer_prefix_object_test.go +++ b/weed/filer/filer_prefix_object_test.go @@ -46,6 +46,50 @@ func TestPromoteToPrefixObject(t *testing.T) { assert.Zero(t, promoted.Attr.TtlSec) } +// TestPrefixObjectKeepsDirectoryAttributes covers an object landing on a +// directory that other keys nest under: the object's data replaces the object +// slot while the directory keeps its own mode and owner — but not fields a +// marker left behind, which belong to the object the marker held. +func TestPrefixObjectKeepsDirectoryAttributes(t *testing.T) { + f, store := newTestFilerWithStubStore(t) + ctx := context.Background() + + dirPath := util.FullPath("/buckets/bkt/a/foo") + require.NoError(t, store.InsertEntry(ctx, &Entry{ + FullPath: dirPath, + Attr: Attr{ + Mode: os.ModeDir | 0o700, + Uid: 1000, + Gid: 1001, + UserName: "dirowner", + }, + Extended: map[string][]byte{s3_constants.SeaweedFSSSES3Key: []byte("stale")}, + })) + + marked := &Entry{ + FullPath: dirPath, + Attr: Attr{ + Mode: os.ModeDir | 0o771, + Uid: filer_pb.OS_UID, + Gid: filer_pb.OS_GID, + }, + Chunks: []*filer_pb.FileChunk{{FileId: "1,02", Size: 3}}, + Extended: map[string][]byte{s3_constants.SeaweedFSPrefixObject: []byte("true")}, + } + require.NoError(t, f.CreateEntry(ctx, marked, nil, false, false, nil, false, f.MaxFilenameLength)) + + found, err := store.FindEntry(ctx, dirPath) + require.NoError(t, err) + require.NotNil(t, found) + assert.Equal(t, os.ModeDir|0o700, found.Attr.Mode, "the directory keeps its own mode") + assert.Equal(t, uint32(1000), found.Attr.Uid) + assert.Equal(t, uint32(1001), found.Attr.Gid) + assert.Equal(t, "dirowner", found.Attr.UserName) + assert.NotContains(t, found.Extended, s3_constants.SeaweedFSSSES3Key, + "the marker's encryption fields must not leak onto the plaintext object") + assert.Equal(t, marked.Chunks, found.Chunks, "the object's data still lands") +} + // TestExpiredDirectoryIsNotDeletedOnRead pins the other half: a TTL an older build // left on a promoted directory must not take the keys under it with it. func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) { @@ -123,4 +167,3 @@ func TestListDirectoryEntriesContextCanceled(t *testing.T) { require.Error(t, err) assert.ErrorContains(t, err, "context canceled") } - diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index a33fc936cc..eaea6a9f0f 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -379,6 +379,7 @@ message Recompute { string demote_key = 7; // if set, stamp demote_value on the prior name_to_key target when it changes bytes demote_value = 8; // value for demote_key string exclude_name = 9; // if set, skip this child when scanning (e.g. a version about to be deleted) + string only_if_current_name = 10; // if set, apply only while the pointer's name_to_key value still equals this } // ObjectTransactionRequest applies an ordered list of mutations atomically with diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 367cb5a6ca..8ba6fcbfd3 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1702,18 +1702,19 @@ func (x *ObjectMutation) GetKeepRemoteObject() bool { // latest version after a specific version is deleted, where the scan must run // under the lock. type Recompute struct { - state protoimpl.MessageState `protogen:"open.v1"` - ScanDir string `protobuf:"bytes,1,opt,name=scan_dir,json=scanDir,proto3" json:"scan_dir,omitempty"` // directory whose direct children are scanned - Descending bool `protobuf:"varint,2,opt,name=descending,proto3" json:"descending,omitempty"` // pick the child that sorts last by name (else first) - CopyExtended map[string]string `protobuf:"bytes,3,rep,name=copy_extended,json=copyExtended,proto3" json:"copy_extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` // pointer extended key -> source extended key on the chosen child - NameToKey string `protobuf:"bytes,4,opt,name=name_to_key,json=nameToKey,proto3" json:"name_to_key,omitempty"` // if set, store the chosen child's name under this pointer key - SizeToKey string `protobuf:"bytes,5,opt,name=size_to_key,json=sizeToKey,proto3" json:"size_to_key,omitempty"` // if set, store the chosen child's FileSize (decimal) under this pointer key - MtimeToKey string `protobuf:"bytes,6,opt,name=mtime_to_key,json=mtimeToKey,proto3" json:"mtime_to_key,omitempty"` // if set, store the chosen child's Mtime (decimal) under this pointer key - DemoteKey string `protobuf:"bytes,7,opt,name=demote_key,json=demoteKey,proto3" json:"demote_key,omitempty"` // if set, stamp demote_value on the prior name_to_key target when it changes - DemoteValue []byte `protobuf:"bytes,8,opt,name=demote_value,json=demoteValue,proto3" json:"demote_value,omitempty"` // value for demote_key - ExcludeName string `protobuf:"bytes,9,opt,name=exclude_name,json=excludeName,proto3" json:"exclude_name,omitempty"` // if set, skip this child when scanning (e.g. a version about to be deleted) - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + ScanDir string `protobuf:"bytes,1,opt,name=scan_dir,json=scanDir,proto3" json:"scan_dir,omitempty"` // directory whose direct children are scanned + Descending bool `protobuf:"varint,2,opt,name=descending,proto3" json:"descending,omitempty"` // pick the child that sorts last by name (else first) + CopyExtended map[string]string `protobuf:"bytes,3,rep,name=copy_extended,json=copyExtended,proto3" json:"copy_extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` // pointer extended key -> source extended key on the chosen child + NameToKey string `protobuf:"bytes,4,opt,name=name_to_key,json=nameToKey,proto3" json:"name_to_key,omitempty"` // if set, store the chosen child's name under this pointer key + SizeToKey string `protobuf:"bytes,5,opt,name=size_to_key,json=sizeToKey,proto3" json:"size_to_key,omitempty"` // if set, store the chosen child's FileSize (decimal) under this pointer key + MtimeToKey string `protobuf:"bytes,6,opt,name=mtime_to_key,json=mtimeToKey,proto3" json:"mtime_to_key,omitempty"` // if set, store the chosen child's Mtime (decimal) under this pointer key + DemoteKey string `protobuf:"bytes,7,opt,name=demote_key,json=demoteKey,proto3" json:"demote_key,omitempty"` // if set, stamp demote_value on the prior name_to_key target when it changes + DemoteValue []byte `protobuf:"bytes,8,opt,name=demote_value,json=demoteValue,proto3" json:"demote_value,omitempty"` // value for demote_key + ExcludeName string `protobuf:"bytes,9,opt,name=exclude_name,json=excludeName,proto3" json:"exclude_name,omitempty"` // if set, skip this child when scanning (e.g. a version about to be deleted) + OnlyIfCurrentName string `protobuf:"bytes,10,opt,name=only_if_current_name,json=onlyIfCurrentName,proto3" json:"only_if_current_name,omitempty"` // if set, apply only while the pointer's name_to_key value still equals this + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Recompute) Reset() { @@ -1809,6 +1810,13 @@ func (x *Recompute) GetExcludeName() string { return "" } +func (x *Recompute) GetOnlyIfCurrentName() string { + if x != nil { + return x.OnlyIfCurrentName + } + return "" +} + // ObjectTransactionRequest applies an ordered list of mutations atomically with // respect to other writers of the same object, by holding the filer's per-path // lock on lock_key for the whole transaction. The optional condition is checked @@ -7508,7 +7516,7 @@ const file_filer_proto_rawDesc = "" + "\n" + "\x06DELETE\x10\x01\x12\x12\n" + "\x0ePATCH_EXTENDED\x10\x02\x12\x14\n" + - "\x10RECOMPUTE_LATEST\x10\x03\"\x9a\x03\n" + + "\x10RECOMPUTE_LATEST\x10\x03\"\xcb\x03\n" + "\tRecompute\x12\x19\n" + "\bscan_dir\x18\x01 \x01(\tR\ascanDir\x12\x1e\n" + "\n" + @@ -7522,7 +7530,9 @@ const file_filer_proto_rawDesc = "" + "\n" + "demote_key\x18\a \x01(\tR\tdemoteKey\x12!\n" + "\fdemote_value\x18\b \x01(\fR\vdemoteValue\x12!\n" + - "\fexclude_name\x18\t \x01(\tR\vexcludeName\x1a?\n" + + "\fexclude_name\x18\t \x01(\tR\vexcludeName\x12/\n" + + "\x14only_if_current_name\x18\n" + + " \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" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index aba514e8be..29d11c7e15 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -1461,6 +1461,13 @@ func (m *Recompute) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.OnlyIfCurrentName) > 0 { + i -= len(m.OnlyIfCurrentName) + copy(dAtA[i:], m.OnlyIfCurrentName) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.OnlyIfCurrentName))) + i-- + dAtA[i] = 0x52 + } if len(m.ExcludeName) > 0 { i -= len(m.ExcludeName) copy(dAtA[i:], m.ExcludeName) @@ -7228,6 +7235,10 @@ func (m *Recompute) SizeVT() (n int) { if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + l = len(m.OnlyIfCurrentName) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -13649,6 +13660,38 @@ func (m *Recompute) UnmarshalVT(dAtA []byte) error { } m.ExcludeName = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 10: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field OnlyIfCurrentName", 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.OnlyIfCurrentName = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 117edab974..306f7f97a7 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -1165,7 +1165,7 @@ func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput // lock: a directory that outlived the object it completed into shares chunks // with it and goes metadata-only; anything else frees the parts' chunks. func (s3a *S3ApiServer) removeUploadDir(bucket, uploadId, object string) s3err.ErrorCode { - completed, err := s3a.uploadCompleted(s3a.bucketDir(bucket), uploadId, object) + completed, err := s3a.uploadCompleted(bucket, uploadId, object) if err != nil { glog.Errorf("bucket %s remove upload %s completed check: %v", bucket, uploadId, err) return s3err.ErrInternalError @@ -1237,10 +1237,11 @@ func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s // uploadCompleted reports whether the upload assembled into an object: the // object entry, or any version file under .versions, still carries the // upload id completion stamps on it. -func (s3a *S3ApiServer) uploadCompleted(bucketDir, uploadId, objectKey string) (bool, error) { +func (s3a *S3ApiServer) uploadCompleted(bucket, uploadId, objectKey string) (bool, error) { if objectKey == "" { return false, nil } + bucketDir := s3a.bucketDir(bucket) name := path.Base(objectKey) dir := path.Dir(objectKey) if dir == "." { @@ -1248,7 +1249,7 @@ func (s3a *S3ApiServer) uploadCompleted(bucketDir, uploadId, objectKey string) ( } objectDir := path.Join(bucketDir, dir) - entry, err := s3a.getEntry(objectDir, name) + entry, err := s3a.getObjectEntryRoutedByKey(bucket, objectKey) if err != nil && !isFilerNotFound(err) { return false, err } @@ -1256,7 +1257,7 @@ func (s3a *S3ApiServer) uploadCompleted(bucketDir, uploadId, objectKey string) ( return true, nil } - versions, _, err := s3a.list(objectDir+"/"+name+s3_constants.VersionsFolder, "", "", false, math.MaxInt32) + versions, _, err := s3a.listObjectScopedEntriesRoutedByKey(bucket, objectKey, objectDir+"/"+name+s3_constants.VersionsFolder, "", "", false, math.MaxInt32) if err != nil { if isFilerNotFound(err) { return false, nil diff --git a/weed/s3api/filer_util_tags.go b/weed/s3api/filer_util_tags.go index ee2f5da33b..373b26a784 100644 --- a/weed/s3api/filer_util_tags.go +++ b/weed/s3api/filer_util_tags.go @@ -4,6 +4,7 @@ import ( "context" "strings" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" ) @@ -34,9 +35,9 @@ func (s3a *S3ApiServer) getTags(parentDirectoryPath string, entryName string) (t return } -func (s3a *S3ApiServer) setTags(parentDirectoryPath string, entryName string, tags map[string]string) (err error) { +func (s3a *S3ApiServer) setTags(owner pb.ServerAddress, parentDirectoryPath string, entryName string, tags map[string]string) (err error) { - return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s3a.withWriteOwnerFilerClient(owner, func(client filer_pb.SeaweedFilerClient) error { resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ Directory: parentDirectoryPath, @@ -70,9 +71,9 @@ func (s3a *S3ApiServer) setTags(parentDirectoryPath string, entryName string, ta } -func (s3a *S3ApiServer) rmTags(parentDirectoryPath string, entryName string) (err error) { +func (s3a *S3ApiServer) rmTags(owner pb.ServerAddress, parentDirectoryPath string, entryName string) (err error) { - return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s3a.withWriteOwnerFilerClient(owner, func(client filer_pb.SeaweedFilerClient) error { resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ Directory: parentDirectoryPath, diff --git a/weed/s3api/s3api_copy_source_classify_test.go b/weed/s3api/s3api_copy_source_classify_test.go index 9eeb3b8a68..609baf32e0 100644 --- a/weed/s3api/s3api_copy_source_classify_test.go +++ b/weed/s3api/s3api_copy_source_classify_test.go @@ -3,15 +3,214 @@ package s3api import ( "errors" "fmt" + "net/http" + "net/http/httptest" + "strings" "testing" + "time" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) +// The gateway's configured filer is a replica, while PUT commits on the key's +// owner. No metadata is replicated in this test, making the visibility gap +// deterministic even though both HTTP handlers run on the same gateway. +func TestCopyAfterRoutedPut(t *testing.T) { + util_http.InitGlobalHttpClient() + for _, versioning := range []string{"", s3_constants.VersioningSuspended} { + for _, multipart := range []bool{false, true} { + t.Run(fmt.Sprintf("versioning=%s/multipart=%t", versioning, multipart), func(t *testing.T) { + volume := startFakeVolumeServer(t) + owner := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + replica := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + ownerAddr := startFakeFiler(t, owner) + replicaAddr := startFakeFiler(t, replica) + s3a := newPutTestServer(t, replicaAddr) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("source", &BucketConfig{Name: "source"}) + s3a.bucketConfigCache.Set("destination", &BucketConfig{Name: "destination"}) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1) + + put := httptest.NewRequest(http.MethodPut, "/source/tmp/object", nil) + put = mux.SetURLVars(put, map[string]string{"bucket": "source", "object": "tmp/object"}) + put.Header.Set("X-Amz-Meta-Revision", "fresh") + putResult := httptest.NewRecorder() + s3a.PutObjectHandler(putResult, put) + require.Equal(t, http.StatusOK, putResult.Code, putResult.Body.String()) + owner.mu.Lock() + stored := owner.entries["/buckets/source/tmp/object"] + owner.mu.Unlock() + require.NotNil(t, stored, "PUT must commit on the owner") + replica.mu.Lock() + replicated := replica.entries["/buckets/source/tmp/object"] + replica.mu.Unlock() + require.Nil(t, replicated, "the replica must still be behind") + require.Equal(t, replicaAddr, s3a.filerClient.GetCurrentFiler()) + + // A regular object remains the null version when versioning is + // suspended. Both copy paths must find it on the write owner. + s3a.bucketConfigCache.Set("source", &BucketConfig{Name: "source", Versioning: versioning}) + copyURL := "/destination/final/object" + var uploadID string + if multipart { + uploadID = s3a.generateUploadID("final/object") + replica.mu.Lock() + replica.entries["/buckets/destination/.uploads/"+uploadID] = &filer_pb.Entry{ + Name: uploadID, IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte("final/object")}, + } + replica.mu.Unlock() + copyURL += "?partNumber=1&uploadId=" + uploadID + } + copyRequest := httptest.NewRequest(http.MethodPut, copyURL, nil) + copyRequest = mux.SetURLVars(copyRequest, map[string]string{"bucket": "destination", "object": "final/object"}) + copyRequest.Header.Set("X-Amz-Copy-Source", "/source/tmp/object") + copyResult := httptest.NewRecorder() + if multipart { + s3a.CopyObjectPartHandler(copyResult, copyRequest) + } else { + s3a.CopyObjectHandler(copyResult, copyRequest) + } + require.Equal(t, http.StatusOK, copyResult.Code, copyResult.Body.String()) + require.Equal(t, replicaAddr, s3a.filerClient.GetCurrentFiler(), "a source read must not change the gateway's default filer") + + replica.mu.Lock() + if multipart { + var part *filer_pb.Entry + for key, entry := range replica.entries { + if strings.HasPrefix(key, "/buckets/destination/.uploads/"+uploadID+"/") && !entry.IsDirectory { + part = entry + } + } + require.NotNil(t, part, "copy must persist the part") + require.Equal(t, stored.Attributes.Md5, part.Attributes.Md5) + replica.mu.Unlock() + } else { + // The destination commits on its write owner, the same filer a + // later copy resolves it from before replication. + require.Nil(t, replica.entries["/buckets/destination/final/object"], "the replica must still be behind") + replica.mu.Unlock() + owner.mu.Lock() + copied := owner.entries["/buckets/destination/final/object"] + owner.mu.Unlock() + require.NotNil(t, copied, "copy must persist the destination on the owner") + require.Equal(t, "fresh", string(copied.Extended["X-Amz-Meta-Revision"])) + require.Equal(t, stored.Attributes.Md5, copied.Attributes.Md5) + + // The destination is immediately usable as a copy source even + // though the replica never saw it. + second := httptest.NewRequest(http.MethodPut, "/destination/final/copy2", nil) + second = mux.SetURLVars(second, map[string]string{"bucket": "destination", "object": "final/copy2"}) + second.Header.Set("X-Amz-Copy-Source", "/destination/final/object") + secondResult := httptest.NewRecorder() + s3a.CopyObjectHandler(secondResult, second) + require.Equal(t, http.StatusOK, secondResult.Code, secondResult.Body.String()) + } + }) + } + } +} + +// A copy into a versioned bucket commits the version file and the .versions +// pointer on the destination's write owner. Reads by versionId, latest-version +// reads, and copy-by-versionId must resolve there before replication. +func TestVersionedCopyDestinationReadsOnOwner(t *testing.T) { + util_http.InitGlobalHttpClient() + volume := startFakeVolumeServer(t) + owner := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + replica := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + ownerAddr := startFakeFiler(t, owner) + replicaAddr := startFakeFiler(t, replica) + s3a := newPutTestServer(t, replicaAddr) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("source", &BucketConfig{Name: "source"}) + s3a.bucketConfigCache.Set("destination", &BucketConfig{Name: "destination", Versioning: s3_constants.VersioningEnabled}) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1) + + put := httptest.NewRequest(http.MethodPut, "/source/tmp/object", nil) + put = mux.SetURLVars(put, map[string]string{"bucket": "source", "object": "tmp/object"}) + putResult := httptest.NewRecorder() + s3a.PutObjectHandler(putResult, put) + require.Equal(t, http.StatusOK, putResult.Code, putResult.Body.String()) + + copyRequest := httptest.NewRequest(http.MethodPut, "/destination/final/object", nil) + copyRequest = mux.SetURLVars(copyRequest, map[string]string{"bucket": "destination", "object": "final/object"}) + copyRequest.Header.Set("X-Amz-Copy-Source", "/source/tmp/object") + copyResult := httptest.NewRecorder() + s3a.CopyObjectHandler(copyResult, copyRequest) + require.Equal(t, http.StatusOK, copyResult.Code, copyResult.Body.String()) + versionId := copyResult.Header().Get("x-amz-version-id") + require.NotEmpty(t, versionId) + + versionKey := "/buckets/destination/final/object.versions/v_" + versionId + owner.mu.Lock() + require.NotNil(t, owner.entries[versionKey], "version file must commit on the owner") + require.NotNil(t, owner.entries["/buckets/destination/final/object.versions"], "the owner must hold the .versions pointer") + owner.mu.Unlock() + replica.mu.Lock() + require.Nil(t, replica.entries[versionKey], "the replica must still be behind") + require.Nil(t, replica.entries["/buckets/destination/final/object.versions"], "the replica must not hold the pointer") + replica.mu.Unlock() + + entry, err := s3a.getSpecificObjectVersion("destination", "final/object", versionId) + require.NoError(t, err) + require.NotNil(t, entry) + + latest, err := s3a.getLatestObjectVersion("destination", "final/object") + require.NoError(t, err) + require.Equal(t, versionId, string(latest.Extended[s3_constants.ExtVersionIdKey])) + + src, err := s3a.resolveCopySourceEntry("destination", "final/object", versionId, s3_constants.VersioningEnabled) + require.NoError(t, err) + require.NotNil(t, src) + + second := httptest.NewRequest(http.MethodPut, "/destination/final/copy2", nil) + second = mux.SetURLVars(second, map[string]string{"bucket": "destination", "object": "final/copy2"}) + second.Header.Set("X-Amz-Copy-Source", "/destination/final/object?versionId="+versionId) + secondResult := httptest.NewRecorder() + s3a.CopyObjectHandler(secondResult, second) + require.Equal(t, http.StatusOK, secondResult.Code, secondResult.Body.String()) +} + +func TestCopySourcePrefersWriteOwner(t *testing.T) { + for _, tc := range []struct { + name string + ownerEntry *filer_pb.Entry + wantCode s3err.ErrorCode + }{ + {name: "overwrite", ownerEntry: &filer_pb.Entry{Name: "object", Content: []byte("fresh")}}, + {name: "deleted", wantCode: s3err.ErrNoSuchKey}, + } { + t.Run(tc.name, func(t *testing.T) { + owner := &fakeLookupFiler{entry: tc.ownerEntry} + replica := &fakeLookupFiler{entry: &filer_pb.Entry{Name: "object", Content: []byte("stale")}} + ownerAddr := startFakeFiler(t, owner) + s3a := newPutTestServer(t, startFakeFiler(t, replica)) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1) + + entry, err := s3a.resolveCopySourceEntry("source", "/tmp/object", "", "") + require.Equal(t, tc.wantCode, classifyCopySourceError(entry, err)) + if tc.wantCode == s3err.ErrNone { + require.Equal(t, "fresh", string(entry.Content)) + } + }) + } +} + func TestClassifyCopySourceError(t *testing.T) { deleteMarker := &filer_pb.Entry{ Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, diff --git a/weed/s3api/s3api_directory_marker.go b/weed/s3api/s3api_directory_marker.go index 0c97382a56..08c5f55603 100644 --- a/weed/s3api/s3api_directory_marker.go +++ b/weed/s3api/s3api_directory_marker.go @@ -9,7 +9,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" - "github.com/seaweedfs/seaweedfs/weed/util" ) // An explicit directory marker - the key "dir/" created by PutObject on a @@ -29,10 +28,10 @@ import ( func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object string) s3err.ErrorCode { governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object) - markerDir := s3a.bucketDir(bucket) + "/" + strings.TrimSuffix(strings.TrimPrefix(object, "/"), "/") - dir, name := util.FullPath(markerDir).DirAndName() + trimmedObject := strings.TrimSuffix(strings.TrimPrefix(object, "/"), "/") + markerDir := s3a.bucketDir(bucket) + "/" + trimmedObject - entry, err := s3a.getEntry(dir, name) + entry, err := s3a.getObjectEntryRoutedByKey(bucket, trimmedObject) switch { case errors.Is(err, filer_pb.ErrNotFound): return s3err.ErrNone // deleting a key that is not there is a success @@ -60,13 +59,13 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object st // Drop a history an older build recorded for this key. Nothing writes one now, and // leaving it behind keeps reporting the key in ListObjectVersions, so a history we // cannot read or remove fails the delete rather than half finishing it. - switch _, historyErr := s3a.getEntry(markerDir, s3_constants.VersionsFolder); { + switch _, historyErr := s3a.getObjectScopedEntryRoutedByKey(bucket, trimmedObject, markerDir, s3_constants.VersionsFolder); { case historyErr == nil: // The removal below takes every entry under the key, so each has to be clear // of a lock of its own. versionsDir := markerDir + "/" + s3_constants.VersionsFolder for startFrom := ""; ; { - entries, isLast, listErr := s3a.list(versionsDir, "", startFrom, false, 1000) + entries, isLast, listErr := s3a.listObjectScopedEntriesRoutedByKey(bucket, trimmedObject, versionsDir, "", startFrom, false, 1000) if listErr != nil { glog.Errorf("deleteDirectoryMarker: cannot list history of %s/%s: %v", bucket, object, listErr) return s3err.ErrInternalError diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index 20e76b62cd..81b42d0501 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -80,10 +80,31 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle } return done(), nil case s3_constants.VersioningSuspended: - // Best-effort null delete; NotFound is benign. - if err := s3a.deleteSpecificObjectVersion(ctx, req.Bucket, req.ObjectPath, "null", metadataOnly); err != nil { - if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrVersionNotFound) { - return retryLater("TRANSPORT_ERROR: deleteNullVersion: " + err.Error()), nil + // Best-effort null delete; NotFound is benign. Route it when an + // owner resolves — on the default filer an absent entry deletes as + // success, and a null version that only reached the owner would + // linger. Lifecycle never bypasses retention. + nullDeleted := false + if owner := s3a.routableWriteOwner(req.Bucket, req.ObjectPath); owner != "" { + worm, lockErr := s3a.isObjectLockEnabled(req.Bucket) + if lockErr == nil { + code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly) + if settled { + if code == s3err.ErrAccessDenied { + return retryLater("ACCESS_DENIED: deleteNullVersion: object lock"), nil + } + if code != s3err.ErrNone { + return retryLater("TRANSPORT_ERROR: routedDeleteNullVersion: " + s3err.GetAPIError(code).Code), nil + } + nullDeleted = true + } + } + } + if !nullDeleted { + if err := s3a.deleteSpecificObjectVersion(ctx, req.Bucket, req.ObjectPath, "null", metadataOnly); err != nil { + if !errors.Is(err, filer_pb.ErrNotFound) && !errors.Is(err, ErrVersionNotFound) { + return retryLater("TRANSPORT_ERROR: deleteNullVersion: " + err.Error()), nil + } } } if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath); err != nil { @@ -137,11 +158,46 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle return outcome, err } } - if err := s3a.deleteSpecificObjectVersion(ctx, req.Bucket, req.ObjectPath, req.VersionId, metadataOnly); err != nil { - if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrObjectNotFound) { - return noopResolved("NOT_FOUND_AT_DELETE"), nil + // Route the version delete when an owner resolves: its recompute + // repoints .versions while the file is removed, all on the filer the + // routed reads consult. A routed failure retries rather than falling + // back, where an absent entry would delete as success while the owner + // still holds the version. Lifecycle never bypasses retention. + versionDeleted := false + if owner := s3a.routableWriteOwner(req.Bucket, req.ObjectPath); owner != "" { + worm, lockErr := s3a.isObjectLockEnabled(req.Bucket) + if lockErr == nil { + // The "null" version is the object entry, not a .versions file. + if req.VersionId == "null" { + code, settled := s3a.routedDeleteNullVersion(owner, req.Bucket, req.ObjectPath, worm, false, metadataOnly) + if settled { + if code == s3err.ErrAccessDenied { + return retryLater("ACCESS_DENIED: deleteSpecificVersion: object lock"), nil + } + if code != s3err.ErrNone { + return retryLater("TRANSPORT_ERROR: routedDeleteNullVersion: " + s3err.GetAPIError(code).Code), nil + } + versionDeleted = true + } + } else { + switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm, false, metadataOnly); code { + case s3err.ErrNone: + versionDeleted = true + case s3err.ErrAccessDenied: + return retryLater("ACCESS_DENIED: deleteSpecificVersion: object lock"), nil + default: + return retryLater("TRANSPORT_ERROR: routedDeleteSpecificVersion: " + s3err.GetAPIError(code).Code), nil + } + } + } + } + if !versionDeleted { + if err := s3a.deleteSpecificObjectVersion(ctx, req.Bucket, req.ObjectPath, req.VersionId, metadataOnly); err != nil { + if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrObjectNotFound) { + return noopResolved("NOT_FOUND_AT_DELETE"), nil + } + return retryLater("TRANSPORT_ERROR: deleteSpecificVersion: " + err.Error()), nil } - return retryLater("TRANSPORT_ERROR: deleteSpecificVersion: " + err.Error()), nil } recordMetadataOnlyIf(metadataOnly, req) return done(), nil diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index ea962a5fb2..f618fd569a 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -20,6 +20,7 @@ import ( "time" "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/remote_pb" "github.com/seaweedfs/seaweedfs/weed/remote_storage" @@ -416,7 +417,6 @@ func (s3a *S3ApiServer) checkDirectoryObject(bucket, object string) (*filer_pb.E return nil, false, nil // Not a directory object } - bucketDir := s3a.bucketDir(bucket) cleanObject := strings.TrimSuffix(object, "/") if cleanObject == "" { @@ -424,7 +424,7 @@ func (s3a *S3ApiServer) checkDirectoryObject(bucket, object string) (*filer_pb.E } // Check if directory exists - dirEntry, err := s3a.getEntry(bucketDir, cleanObject) + dirEntry, err := s3a.getObjectEntryRoutedByKey(bucket, cleanObject) if err != nil { if errors.Is(err, filer_pb.ErrNotFound) { return nil, true, nil // Directory object requested but doesn't exist @@ -462,8 +462,7 @@ func (s3a *S3ApiServer) resolveObjectEntry(bucket, object, versionId string) (*f } // For non-versioned buckets, verify directly - bucketDir := s3a.bucketDir(bucket) - return s3a.getEntry(bucketDir, object) + return s3a.getObjectEntryRoutedByKey(bucket, object) } // serveDirectoryContent serves the content of a directory object directly @@ -961,7 +960,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) versionsDir := normalizedObject + s3_constants.VersionsFolder // Quick check (no retries) for .versions/ directory - versionsEntry, versionsErr := s3a.getEntry(bucketDir, versionsDir) + versionsEntry, versionsErr := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsDir) if versionsErr == nil && versionsEntry != nil { // .versions/ exists, meaning real versions are stored there @@ -974,7 +973,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) } } else if errors.Is(versionsErr, filer_pb.ErrNotFound) { // .versions/ doesn't exist (confirmed not found), check regular path for null version - regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) // A physical parent or bare-key object is not a trailing-slash null marker. if regularErr == nil && isNullVersionObject(regularEntry, normalizedObject) { // Found object at regular path - this is the null version @@ -1251,7 +1250,10 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R return newStreamErrorWithResponse(fmt.Errorf("cached entry authorization failed: %v", code)) } } - if cacheErr == nil && cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { + // Swap only a fill of the resolved generation with matching size; + // totalSize, offset and size were computed from the first entry. + if cacheErr == nil && cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 && + sameRemoteObject(entry, cachedEntry) && int64(filer.FileSize(cachedEntry)) == totalSize { // Cache completed, use cached chunks chunks = cachedEntry.GetChunks() entry = cachedEntry @@ -2656,7 +2658,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request versionsDir := normalizedObject + s3_constants.VersionsFolder // Quick check (no retries) for .versions/ directory - versionsEntry, versionsErr := s3a.getEntry(bucketDir, versionsDir) + versionsEntry, versionsErr := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsDir) if versionsErr == nil && versionsEntry != nil { // .versions/ exists, meaning real versions are stored there @@ -2669,7 +2671,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request } } else if errors.Is(versionsErr, filer_pb.ErrNotFound) { // .versions/ doesn't exist (confirmed not found), check regular path for null version - regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) // A physical parent or bare-key object is not a trailing-slash null marker. if regularErr == nil && isNullVersionObject(regularEntry, normalizedObject) { // Found object at regular path - this is the null version @@ -3631,6 +3633,30 @@ func (s3a *S3ApiServer) doCacheRemoteObject(ctx context.Context, dir, name strin return cachedEntry, err } +// doCacheRemoteObjectOnFiler is doCacheRemoteObject against one filer: the +// entry being cached may live only on the write owner that served the lookup, +// ahead of peer replication. Callers fall back to doCacheRemoteObject on error. +func (s3a *S3ApiServer) doCacheRemoteObjectOnFiler(ctx context.Context, filer pb.ServerAddress, dir, name string) (*filer_pb.Entry, error) { + if filer == "" { + return s3a.doCacheRemoteObject(ctx, dir, name) + } + var cachedEntry *filer_pb.Entry + err := pb.WithFilerClient(false, 0, filer, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + resp, cacheErr := client.CacheRemoteObjectToLocalCluster(ctx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{ + Directory: dir, + Name: name, + }) + if cacheErr != nil { + return cacheErr + } + if resp != nil && resp.Entry != nil { + cachedEntry = resp.Entry + } + return nil + }) + return cachedEntry, err +} + func (s3a *S3ApiServer) buildVersionedRemoteObjectPath(bucket, object, versionId string) (dir, name string) { // versionId names a .versions file, so a value that is not a valid path // segment falls back to the unversioned remote path rather than escaping it. @@ -3648,6 +3674,19 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool { return entry != nil && (len(entry.GetChunks()) > 0 || len(entry.Content) > 0) } +// sameRemoteObject reports whether two entries describe the same remote object +// generation. A cache response can come back from a filer holding an older +// generation of the key; it must not replace the resolved entry. +func sameRemoteObject(src, cached *filer_pb.Entry) bool { + srcRemote, cachedRemote := src.GetRemoteEntry(), cached.GetRemoteEntry() + if srcRemote == nil || cachedRemote == nil { + return srcRemote == cachedRemote + } + return srcRemote.RemoteETag == cachedRemote.RemoteETag && + srcRemote.RemoteMtime == cachedRemote.RemoteMtime && + srcRemote.RemoteSize == cachedRemote.RemoteSize +} + // remoteCacheStreamingTimeoutNS bounds (nanoseconds) how long the streaming read // path waits for the filer to finish caching a remote-only object. Without a // bound a multi-GB download outlasts the client's read timeout, so the request @@ -3693,7 +3732,10 @@ func (s3a *S3ApiServer) cacheRemoteObjectForStreamingWithShortTimeout(r *http.Re glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: polling cache status for %s/%s (timeout=%v)", dir, name, pollTimeout) - cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) + cachedEntry, err := s3a.doCacheRemoteObjectOnFiler(cacheCtx, s3a.routableWriteOwner(bucket, object), dir, name) + if err != nil { + cachedEntry, err = s3a.doCacheRemoteObject(cacheCtx, dir, name) + } if err != nil { // Distinguish transient errors (timeout/cancellation) from permanent errors if cacheCtx.Err() != nil { @@ -3726,7 +3768,10 @@ func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *fi glog.V(1).Infof("cacheRemoteObjectForStreaming: caching %s/%s (remote size: %d, versionId: %s)", dir, name, entry.RemoteEntry.RemoteSize, versionId) - cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) + cachedEntry, err := s3a.doCacheRemoteObjectOnFiler(cacheCtx, s3a.routableWriteOwner(bucket, object), dir, name) + if err != nil { + cachedEntry, err = s3a.doCacheRemoteObject(cacheCtx, dir, name) + } if err != nil { // A bounded-wait timeout (or client disconnect) is not a cache failure: // the filer keeps downloading and the caller maps a nil return to 503. @@ -3738,7 +3783,7 @@ func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *fi return nil } - if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { + if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 && sameRemoteObject(entry, cachedEntry) { glog.V(1).Infof("cacheRemoteObjectForStreaming: successfully cached %s/%s (%d chunks)", dir, name, len(cachedEntry.GetChunks())) return cachedEntry } @@ -3749,8 +3794,9 @@ func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *fi // cacheRemoteObjectForCopy caches a remote-only source before CopyObject / // CopyObjectPart reads it; otherwise the copy would write a destination with // FileSize > 0 but no chunks/content. Bounded so a stuck cache can't hang -// the copy. Returns nil if caching failed or produced no local data. -func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, object, versionId string) *filer_pb.Entry { +// the copy. Returns nil if caching failed, produced no local data, or the +// fill resolved a different generation than the authorized source. +func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, src *filer_pb.Entry, bucket, object, versionId string) *filer_pb.Entry { const cacheTimeout = 30 * time.Second cacheCtx, cancel := context.WithTimeout(ctx, cacheTimeout) defer cancel() @@ -3759,7 +3805,12 @@ func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, ob glog.V(1).Infof("cacheRemoteObjectForCopy: caching %s/%s (versionId: %s)", dir, name, versionId) - cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) + // An owner-routed source read can return a remote-only entry that has not + // replicated to the default filer; cache it on the filer that has it. + cachedEntry, err := s3a.doCacheRemoteObjectOnFiler(cacheCtx, s3a.routableWriteOwner(bucket, object), dir, name) + if err != nil { + cachedEntry, err = s3a.doCacheRemoteObject(cacheCtx, dir, name) + } if err != nil { if errors.Is(err, context.DeadlineExceeded) { glog.Warningf("cacheRemoteObjectForCopy: timeout caching %s/%s after %v", dir, name, cacheTimeout) @@ -3769,7 +3820,11 @@ func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, ob return nil } - if cachedEntryHasLocalData(cachedEntry) { + // A fallback filer can answer for an older generation of the key; only a + // fill of the resolved source — same remote generation and size — may + // replace it. + if cachedEntryHasLocalData(cachedEntry) && sameRemoteObject(src, cachedEntry) && + filer.FileSize(cachedEntry) == filer.FileSize(src) { glog.V(1).Infof("cacheRemoteObjectForCopy: successfully cached %s/%s (chunks=%d, inline=%d)", dir, name, len(cachedEntry.GetChunks()), len(cachedEntry.Content)) return cachedEntry } @@ -3800,7 +3855,10 @@ func (s3a *S3ApiServer) startBackgroundRemoteCache(bucket, object, versionId str glog.V(2).Infof("startBackgroundRemoteCache: caching disabled for %s/%s", dir, name) return } - _, err := s3a.doCacheRemoteObject(bgCtx, dir, name) + _, err := s3a.doCacheRemoteObjectOnFiler(bgCtx, s3a.routableWriteOwner(bucket, object), dir, name) + if err != nil { + _, err = s3a.doCacheRemoteObject(bgCtx, dir, name) + } if err != nil { glog.V(2).Infof("startBackgroundRemoteCache: cache failed for %s/%s: %v", bucket, object, err) } else { diff --git a/weed/s3api/s3api_object_handlers_attributes.go b/weed/s3api/s3api_object_handlers_attributes.go index 9e8881ebc1..1883c4431a 100644 --- a/weed/s3api/s3api_object_handlers_attributes.go +++ b/weed/s3api/s3api_object_handlers_attributes.go @@ -165,7 +165,7 @@ func (s3a *S3ApiServer) GetObjectAttributesHandler(w http.ResponseWriter, r *htt normalizedObject := s3_constants.NormalizeObjectKey(object) versionsDir := normalizedObject + s3_constants.VersionsFolder - versionsEntry, versionsErr := s3a.getEntry(bucketDir, versionsDir) + versionsEntry, versionsErr := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsDir) if versionsErr == nil && versionsEntry != nil { entry, err = s3a.getLatestObjectVersion(bucket, object) if err != nil { @@ -173,7 +173,7 @@ func (s3a *S3ApiServer) GetObjectAttributesHandler(w http.ResponseWriter, r *htt return } } else if errors.Is(versionsErr, filer_pb.ErrNotFound) { - regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) if regularErr == nil && regularEntry != nil { entry = regularEntry targetVersionId = "null" diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 555dcaf696..69ae416f4e 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -24,6 +24,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" @@ -268,7 +269,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request // writes a destination with FileSize > 0 but no chunks/content. if entry.IsInRemoteOnly() { cacheVersionId := resolvedSourceVersionId(srcVersionId, entry) - cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), srcBucket, srcObject, cacheVersionId) + cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), entry, srcBucket, srcObject, cacheVersionId) if cachedEntry == nil { glog.Errorf("CopyObjectHandler: failed to cache remote-only source %s/%s (version %q)", srcBucket, srcObject, cacheVersionId) w.Header().Set("Retry-After", "5") @@ -728,24 +729,34 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion versionFileName := s3a.getVersionFileName(versionId) versionObjectPath := normalizedObject + s3_constants.VersionsFolder + "/" + versionFileName bucketDir := s3a.bucketDir(dstBucket) + owner := s3a.objectWriteOwner(dstBucket, normalizedObject) - if err = s3a.mkFile(bucketDir, versionObjectPath, dstEntry.Chunks, func(entry *filer_pb.Entry) { - copyEntryToTarget(entry, dstEntry) - }); err != nil { + // The finalize recomputes the latest pointer by scanning .versions on + // the owner, so the version file must commit there too. + 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) + } else { + err = s3a.mkFile(bucketDir, versionObjectPath, dstEntry.Chunks, func(entry *filer_pb.Entry) { + copyEntryToTarget(entry, dstEntry) + }) + } + if err != nil { return "", "", err } // Route the pointer flip to the owner filer when known (off the // distributed lock); RECOMPUTE_LATEST picks the just-written version. - if owner := s3a.objectWriteOwner(dstBucket, normalizedObject); owner != "" { + if owner != "" { if code := s3a.routedVersionedFinalize(owner, dstBucket, normalizedObject, isNewFormatVersionId(versionId)); code != s3err.ErrNone { - if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil { + if rollbackErr := s3a.rollbackCopyVersion(owner, dstBucket, normalizedObject, versionObjectPath); rollbackErr != nil { glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, dstBucket, normalizedObject, rollbackErr) } return "", "", fmt.Errorf("routed finalize for %s/%s: code %d", dstBucket, normalizedObject, code) } } else if err = s3a.updateLatestVersionInDirectory(dstBucket, normalizedObject, versionId, versionFileName, dstEntry); err != nil { - if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil { + if rollbackErr := s3a.rollbackCopyVersion("", dstBucket, normalizedObject, versionObjectPath); rollbackErr != nil { glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, dstBucket, normalizedObject, rollbackErr) } glog.Errorf("CopyObjectHandler: failed to update latest version in directory: %v", err) @@ -759,14 +770,15 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion dstEntry.Extended[s3_constants.ExtVersionIdKey] = []byte("null") dstEntry.Extended[s3_constants.ExtETagKey] = []byte(etag) - if err = s3a.mkFile(dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) { + 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) - }); err != nil { + }, nil); err != nil { return "", "", err } - // mkFile writes through the default filer, so the ownership check reads there too. - if err = s3a.finalizeSuspendedNullWrite("", dstBucket, normalizedObject, s3_constants.ExtETagKey, etag); err != nil { + // The ownership check reads the entry back where the write committed. + if err = s3a.finalizeSuspendedNullWrite(owner, dstBucket, normalizedObject, s3_constants.ExtETagKey, etag); err != nil { glog.Warningf("CopyObjectHandler: failed to retire the null delete marker for %s/%s: %v", dstBucket, normalizedObject, err) } @@ -776,9 +788,12 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion cleanupVersioningMetadata(dstEntry.Extended) dstEntry.Extended[s3_constants.ExtETagKey] = []byte(etag) - if err = s3a.mkFile(dstDir, dstName, dstEntry.Chunks, func(entry *filer_pb.Entry) { + // Copy source reads resolve on the key's write owner, so a copy + // destination must commit there too — a destination written to the + // 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) - }); err != nil { + }, nil); err != nil { return "", "", err } @@ -786,10 +801,29 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion } } -func (s3a *S3ApiServer) rollbackCopyVersion(bucketDir, versionObjectPath string) error { - versionPath := util.FullPath(fmt.Sprintf("%s/%s", bucketDir, versionObjectPath)) +func (s3a *S3ApiServer) rollbackCopyVersion(owner pb.ServerAddress, bucket, normalizedObject, versionObjectPath string) error { + versionPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), versionObjectPath)) versionDir, versionName := versionPath.DirAndName() - return s3a.rmObject(context.Background(), versionDir, versionName, true, false) + if owner == "" { + return s3a.rmObject(context.Background(), versionDir, versionName, true, false) + } + resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ + LockKey: string(versionPath), + RouteKey: s3a.objectRouteKey(bucket, normalizedObject), + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_DELETE, + Directory: versionDir, + Name: versionName, + IsDeleteData: true, + }}, + }) + if err != nil { + return err + } + if resp.Error != "" { + return errors.New(resp.Error) + } + return nil } func (s3a *S3ApiServer) resolveCopySourceEntry(bucket, object, versionId, versioningState string) (*filer_pb.Entry, error) { @@ -805,9 +839,9 @@ func (s3a *S3ApiServer) resolveCopySourceEntry(bucket, object, versionId, versio case s3_constants.VersioningSuspended: return s3a.resolveSuspendedCopySourceEntry(bucket, normalizedObject, "CopyObject") default: - srcPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), normalizedObject)) - dir, name := srcPath.DirAndName() - return s3a.getEntry(dir, name) + // PUT commits on the key's owner, which can differ from this gateway's + // current filer. Read there before peer metadata replication catches up. + return s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) } } @@ -852,9 +886,7 @@ func isManagedCopyMetadataKey(key string) bool { } func (s3a *S3ApiServer) resolveSuspendedCopySourceEntry(bucket, normalizedObject, operation string) (*filer_pb.Entry, error) { - srcPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), normalizedObject)) - dir, name := srcPath.DirAndName() - entry, err := s3a.getEntry(dir, name) + entry, err := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) if err == nil { return entry, nil } @@ -1065,24 +1097,8 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req return } - // Get the source entry with version awareness based on versioning state - var entry *filer_pb.Entry - if srcVersionId != "" { - // Specific version requested - always use version-aware retrieval - entry, err = s3a.getSpecificObjectVersion(srcBucket, srcObject, srcVersionId) - } else if srcVersioningState == s3_constants.VersioningEnabled { - // Versioning enabled - get latest version from .versions directory - entry, err = s3a.getLatestObjectVersion(srcBucket, srcObject) - } else if srcVersioningState == s3_constants.VersioningSuspended { - // Versioning suspended - current object is stored as regular file ("null" version) - // Try regular file first, fall back to latest version if needed - entry, err = s3a.resolveSuspendedCopySourceEntry(srcBucket, s3_constants.NormalizeObjectKey(srcObject), "CopyObjectPart") - } else { - // No versioning configured - use regular retrieval - srcPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(srcBucket), srcObject)) - dir, name := srcPath.DirAndName() - entry, err = s3a.getEntry(dir, name) - } + // Share CopyObject's source routing and version resolution. + entry, err := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState) if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) @@ -1098,7 +1114,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req // iterate entry.GetChunks() and would otherwise produce an empty part. if entry.IsInRemoteOnly() { cacheVersionId := resolvedSourceVersionId(srcVersionId, entry) - cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), srcBucket, srcObject, cacheVersionId) + cachedEntry := s3a.cacheRemoteObjectForCopy(r.Context(), entry, srcBucket, srcObject, cacheVersionId) if cachedEntry == nil { glog.Errorf("CopyObjectPartHandler: failed to cache remote-only source %s/%s (version %q)", srcBucket, srcObject, cacheVersionId) w.Header().Set("Retry-After", "5") diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index caa0069624..9062e7ccc5 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -62,13 +62,12 @@ func (s3a *S3ApiServer) resolveDeleteConditionalEntry(bucket, object, versionId, return nil, errInvalidVersionID } normalizedObject := s3_constants.NormalizeObjectKey(object) - bucketDir := s3a.bucketDir(bucket) if versionId != "" { if versionId == "null" { - return s3a.getEntry(bucketDir, normalizedObject) + return s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) } - return s3a.getEntry(s3a.getVersionedObjectDir(bucket, normalizedObject), s3a.getVersionFileName(versionId)) + return s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, s3a.getVersionedObjectDir(bucket, normalizedObject), s3a.getVersionFileName(versionId)) } switch versioningState { @@ -166,9 +165,28 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v glog.V(2).Infof("deleteVersionedObject: object lock check failed for %s/%s null version: %v", bucket, object, err) return result, s3err.ErrAccessDenied } - if err := s3a.deleteSpecificObjectVersion(r.Context(), bucket, object, "null", false); err != nil { - glog.Errorf("deleteVersionedObject: failed to delete null version for %s/%s: %v", bucket, object, err) - return result, s3err.ErrInternalError + // Delete the null version on the owner the routed reads consult: on + // the default filer an absent entry is a successful delete, so a null + // version that only reached the owner would survive a reported + // success. An unsettled routed attempt falls back to the lock path. + nullDeleted := false + if owner := s3a.routableWriteOwner(bucket, object); owner != "" { + worm, lockErr := s3a.isObjectLockEnabled(bucket) + if lockErr == nil { + code, settled := s3a.routedDeleteNullVersion(owner, bucket, object, worm, worm && governanceBypassAllowed, false) + if settled { + if code != s3err.ErrNone { + return result, code + } + nullDeleted = true + } + } + } + if !nullDeleted { + if err := s3a.deleteSpecificObjectVersion(r.Context(), bucket, object, "null", false); err != nil { + glog.Errorf("deleteVersionedObject: failed to delete null version for %s/%s: %v", bucket, object, err) + return result, s3err.ErrInternalError + } } // Suspended versioning overwrites the null version with a single null delete // marker (S3 spec), so the marker replaces any prior one instead of piling up. @@ -317,9 +335,9 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque } } if versionId == "null" { - deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass) + deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass, false) } else { - deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass), true + deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass, false), true } } } diff --git a/weed/s3api/s3api_object_handlers_extract.go b/weed/s3api/s3api_object_handlers_extract.go index 1bdcb523d9..d57884d361 100644 --- a/weed/s3api/s3api_object_handlers_extract.go +++ b/weed/s3api/s3api_object_handlers_extract.go @@ -250,6 +250,9 @@ func (s3a *S3ApiServer) extractZipObject(w http.ResponseWriter, r *http.Request, if !objectLockEnabled { return s3err.ErrNone } + // mkdir replaces the entry on the default filer, so the lock check + // must read there too — a routed read can miss a marker the owner + // has not received yet. existing, existErr := s3a.getEntry(dirName, entryName) if existErr != nil { return s3err.ErrNone diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 80d7efbacb..abe4e98607 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -249,6 +249,9 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) if !objectLockEnabled { return s3err.ErrNone } + // mkdir replaces the entry on the default filer, so the lock check + // must read there too — a routed read can miss a marker the owner + // has not received yet. existing, existErr := s3a.getEntry(fullDirPath, entryName) if existErr != nil { return s3err.ErrNone diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 0ef0426fac..0d08dcdb7c 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -11,6 +11,7 @@ import ( "net/http/httptest" "path" "slices" + "strconv" "strings" "sync" "testing" @@ -296,10 +297,16 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O } if !f.txnNoApply { for _, m := range req.Mutations { - if m.Type == filer_pb.ObjectMutation_PUT { + switch m.Type { + case filer_pb.ObjectMutation_PUT: entry := proto.Clone(m.Entry).(*filer_pb.Entry) filer_pb.BeforeEntrySerialization(entry.Chunks) f.entries[m.Directory+"/"+m.Entry.Name] = entry + f.ensureParents(m.Directory) + case filer_pb.ObjectMutation_DELETE: + delete(f.entries, m.Directory+"/"+m.Name) + case filer_pb.ObjectMutation_RECOMPUTE_LATEST: + f.applyRecomputeLatest(m) } } } @@ -310,6 +317,99 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O return &filer_pb.ObjectTransactionResponse{}, f.txnErr } +// ensureParents mirrors the filer's InsertEntry, which creates missing parent +// directory entries for a stored object (e.g. the .versions dir a version file +// lands under). +func (f *ambiguousPutFiler) ensureParents(dir string) { + for dir != "" && dir != "/" && dir != "." { + parent, name := path.Dir(dir), path.Base(dir) + if _, exists := f.entries[parent+"/"+name]; exists { + return + } + f.entries[parent+"/"+name] = &filer_pb.Entry{Name: name, IsDirectory: true} + dir = parent + } +} + +// applyRecomputeLatest mirrors FilerServer.applyRecomputeLatest: re-derive the +// pointer entry from the children of recompute.scan_dir, copying the mapped +// extended keys off the chosen child. +func (f *ambiguousPutFiler) applyRecomputeLatest(m *filer_pb.ObjectMutation) { + rc := m.Recompute + pointerKey := m.Directory + "/" + m.Name + pointer := f.entries[pointerKey] + if pointer == nil { + return + } + if pointer.Extended == nil { + pointer.Extended = make(map[string][]byte) + } + priorName := string(pointer.Extended[rc.NameToKey]) + + var names []string + for key := range f.entries { + if path.Dir(key) == rc.ScanDir { + names = append(names, path.Base(key)) + } + } + slices.Sort(names) + var chosen *filer_pb.Entry + pick := func(i int) *filer_pb.Entry { + return f.entries[rc.ScanDir+"/"+names[i]] + } + if rc.Descending { + for i := len(names) - 1; i >= 0; i-- { + if names[i] != rc.ExcludeName { + chosen = pick(i) + break + } + } + } else { + for i := 0; i < len(names); i++ { + if names[i] != rc.ExcludeName { + chosen = pick(i) + break + } + } + } + + if chosen == nil { + for pointerKey := range rc.CopyExtended { + delete(pointer.Extended, pointerKey) + } + for _, k := range []string{rc.NameToKey, rc.SizeToKey, rc.MtimeToKey} { + if k != "" { + delete(pointer.Extended, k) + } + } + } else { + for pointerKey, sourceKey := range rc.CopyExtended { + if v, ok := chosen.Extended[sourceKey]; ok { + pointer.Extended[pointerKey] = v + } else { + delete(pointer.Extended, pointerKey) + } + } + if rc.NameToKey != "" { + pointer.Extended[rc.NameToKey] = []byte(chosen.Name) + } + if rc.SizeToKey != "" && chosen.Attributes != nil { + pointer.Extended[rc.SizeToKey] = []byte(strconv.FormatUint(chosen.Attributes.FileSize, 10)) + } + if rc.MtimeToKey != "" && chosen.Attributes != nil { + pointer.Extended[rc.MtimeToKey] = []byte(strconv.FormatInt(chosen.Attributes.Mtime, 10)) + } + if rc.DemoteKey != "" && priorName != "" && priorName != chosen.Name { + if prior := f.entries[rc.ScanDir+"/"+priorName]; prior != nil { + if prior.Extended == nil { + prior.Extended = make(map[string][]byte) + } + prior.Extended[rc.DemoteKey] = rc.DemoteValue + } + } + } +} + func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { f.mu.Lock() defer f.mu.Unlock() diff --git a/weed/s3api/s3api_object_handlers_rename.go b/weed/s3api/s3api_object_handlers_rename.go index fc1ba4436e..4ad6286691 100644 --- a/weed/s3api/s3api_object_handlers_rename.go +++ b/weed/s3api/s3api_object_handlers_rename.go @@ -106,7 +106,10 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque renameSource := r.Header.Get(s3_constants.AmzRenameSource) errCode = s3a.withRenameWriteLocks(bucket, srcObject, dstObject, func() s3err.ErrorCode { - entry, err := s3a.resolveCopySourceEntry(bucket, srcObject, "", "") + // The move runs on the default filer, so the source is read there too — + // a routed read could miss an entry the rename can still reach. + srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, srcObject)).DirAndName() + entry, err := s3a.getEntry(srcDir, srcName) srcIsPrefixObject := entry.IsPrefixObject() entry = prefixObjectSource(entry) srcErrCode := classifyCopySourceError(entry, err) @@ -212,7 +215,9 @@ func (s3a *S3ApiServer) pickRenameSource(bucket string, candidates []string) str if strings.HasSuffix(candidate, "/") { continue } - if !renameSourceAbsent(s3a.resolveCopySourceEntry(bucket, candidate, "", "")) { + srcDir, srcName := util.FullPath(s3a.toFilerPath(bucket, candidate)).DirAndName() + entry, err := s3a.getEntry(srcDir, srcName) + if !renameSourceAbsent(entry, err) { return candidate } } @@ -400,6 +405,9 @@ func (s3a *S3ApiServer) withRenameWriteLocks(bucket, srcObject, dstObject string func (s3a *S3ApiServer) lookupRenameDestination(bucket, dstObject string) (*filer_pb.Entry, s3err.ErrorCode) { dstDir, dstName := util.FullPath(s3a.toFilerPath(bucket, dstObject)).DirAndName() + // AtomicRenameEntry and the retry token it carries land on the default + // filer, so the destination is read there too — a routed read can miss + // both until replication. entry, err := s3a.getEntry(dstDir, dstName) if err != nil { if errors.Is(err, filer_pb.ErrNotFound) { diff --git a/weed/s3api/s3api_object_handlers_tagging.go b/weed/s3api/s3api_object_handlers_tagging.go index c55b38c09c..850de45629 100644 --- a/weed/s3api/s3api_object_handlers_tagging.go +++ b/weed/s3api/s3api_object_handlers_tagging.go @@ -75,10 +75,7 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R } } else { // Handle regular (non-versioned) object tagging retrieval - target := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), object)) - dir, name := target.DirAndName() - - entry, err = s3a.getEntry(dir, name) + entry, err = s3a.getObjectEntryRoutedByKey(bucket, object) if err != nil { if err == filer_pb.ErrNotFound { glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err) @@ -212,7 +209,7 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R return } - if err = s3a.setTags(dir, name, tags); err != nil { + if err = s3a.setTags(s3a.routableWriteOwner(bucket, object), dir, name, tags); err != nil { if err == filer_pb.ErrNotFound { glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err) s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) @@ -256,8 +253,8 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R entry.Extended[S3TAG_PREFIX+k] = []byte(v) } - // Update the entry with tags - err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + // Update the entry with tags, on the owner the read resolved it from + err = s3a.withWriteOwnerFilerClient(s3a.routableWriteOwner(bucket, object), func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.UpdateEntryRequest{ Directory: updateDirectory, Entry: entry, @@ -352,7 +349,7 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt return } - err := s3a.rmTags(dir, name) + err := s3a.rmTags(s3a.routableWriteOwner(bucket, object), dir, name) if err != nil { if err == filer_pb.ErrNotFound { glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err) @@ -399,8 +396,8 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt return } - // Update the entry - err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + // Update the entry on the owner the read resolved it from + err = s3a.withWriteOwnerFilerClient(s3a.routableWriteOwner(bucket, object), func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.UpdateEntryRequest{ Directory: updateDirectory, Entry: entry, diff --git a/weed/s3api/s3api_object_routed_read.go b/weed/s3api/s3api_object_routed_read.go index f267478794..ae2c4dd74a 100644 --- a/weed/s3api/s3api_object_routed_read.go +++ b/weed/s3api/s3api_object_routed_read.go @@ -21,12 +21,18 @@ const unreachableOwnerTTL = 2 * time.Second // probes the key's prior owner once during a rebalance window; falls back to // getEntry when no owner is resolvable. func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer_pb.Entry, error) { - fullPath := util.NewFullPath(s3a.bucketDir(bucket), object) + dir, name := util.NewFullPath(s3a.bucketDir(bucket), object).DirAndName() + return s3a.getObjectScopedEntryRoutedByKey(bucket, object, dir, name) +} +// getObjectScopedEntryRoutedByKey reads dir/name the way +// getObjectEntryRoutedByKey reads the object itself, for the entries an +// object's writes commit on its owner: its .versions directory and the version +// files under it. bucket/object picks the owner; dir/name picks the entry. +func (s3a *S3ApiServer) getObjectScopedEntryRoutedByKey(bucket, object, dir, name string) (*filer_pb.Entry, error) { owner := s3a.routableWriteOwner(bucket, object) if owner == "" || s3a.filerClient == nil { - entry, _, _, err := filer_pb.GetEntry(context.Background(), s3a, fullPath) - return entry, err + return s3a.getEntry(dir, name) } // Skip an owner whose recent read hit a transport error; read local-first until @@ -36,7 +42,6 @@ func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer preferred = "" } - dir, name := fullPath.DirAndName() var entry *filer_pb.Entry err := s3a.withFilerClientFailover(context.Background(), preferred, false, true, func(client filer_pb.SeaweedFilerClient) error { resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ @@ -62,6 +67,37 @@ func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer return entry, err } +// listObjectScopedEntriesRoutedByKey lists parentDirectoryPath preferring +// bucket/object's write owner, for scans of directories that commit under the +// object's owner (.versions). Mirrors s3a.list when no owner is resolvable. +func (s3a *S3ApiServer) listObjectScopedEntriesRoutedByKey(bucket, object, parentDirectoryPath, prefix, startFrom string, inclusive bool, limit uint32) (entries []*filer_pb.Entry, isLast bool, err error) { + owner := s3a.routableWriteOwner(bucket, object) + if owner == "" || s3a.filerClient == nil { + return s3a.list(parentDirectoryPath, prefix, startFrom, inclusive, limit) + } + preferred := owner + if s3a.ownerRecentlyUnreachable(owner) { + preferred = "" + } + err = listRetry(parentDirectoryPath, func() error { + entries = entries[:0] + isLast = false + return s3a.withFilerClientFailover(context.Background(), preferred, false, true, func(client filer_pb.SeaweedFilerClient) error { + return filer_pb.SeaweedList(context.Background(), client, parentDirectoryPath, prefix, func(entry *filer_pb.Entry, isLastEntry bool) error { + entries = append(entries, entry) + if isLastEntry { + isLast = true + } + return nil + }, startFrom, inclusive, limit) + }) + }) + if len(entries) == 0 { + isLast = true + } + return entries, isLast, err +} + func (s3a *S3ApiServer) priorWriteOwner(bucket, object string) pb.ServerAddress { if object == "" || s3a.objectWriteLockClient == nil { return "" diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 27c5188423..be70ed1fcb 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -29,7 +29,9 @@ const objectWriteRouteKeyPrefix = s3_constants.ObjectWriteRouteKeyPrefix // filer (reached because the gateway's ring view was stale) forwards the // transaction to the owner. All of an object's writes share this key. func (s3a *S3ApiServer) objectRouteKey(bucket, object string) string { - return objectWriteRouteKeyPrefix + s3a.toFilerPath(bucket, object) + // A trailing-slash marker key ("dir/") names the same entry as the object + // ("dir"), so both hash and lock to one owner. + return objectWriteRouteKeyPrefix + strings.TrimSuffix(s3a.toFilerPath(bucket, object), "/") } // routableWriteOwner returns the owner filer for an object's writes, or "" to @@ -187,6 +189,27 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O return resp, err } +// withWriteOwnerFilerClient runs fn on the object's write owner so a read +// routed to the owner sees the change without waiting for replication. An +// error fn returns on the owner is authoritative and surfaces as-is: replaying +// the mutation on a lagging replica could update a stale entry and still +// report success. Only a channel that never came up — provably nothing sent — +// lets the failover walk reach another filer; with no resolved owner the +// default filer applies the mutation. +func (s3a *S3ApiServer) withWriteOwnerFilerClient(owner pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error { + if owner == "" { + return s3a.WithFilerClient(false, fn) + } + if s3a.filerClient == nil { + return pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, fn) + } + preferred := owner + if s3a.ownerRecentlyUnreachable(owner) { + preferred = "" + } + return s3a.withFilerClientFailover(context.Background(), preferred, false, false, fn) +} + // routedPut writes an entry as a PUT, optionally followed by finalize mutations, // as one ObjectTransaction applied in order under lockKey on the owner filer. // lockKey is normally the entry's own path; a versioned add instead passes the @@ -242,6 +265,15 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir if err != nil { return err } + if resp.ErrorCode == filer_pb.FilerError_EXISTING_IS_DIRECTORY && !isReservedDirectoryName(name) { + // 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) + if err != nil { + return err + } + } if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { return errUploadRemoved } diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 470a63e0f3..0f23779d72 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -258,6 +258,7 @@ type fakeTxnFiler struct { filer_pb.UnimplementedSeaweedFilerServer calls int32 lastReq *filer_pb.ObjectTransactionRequest + reqs []*filer_pb.ObjectTransactionRequest resp *filer_pb.ObjectTransactionResponse err error } @@ -265,6 +266,7 @@ type fakeTxnFiler struct { func (f *fakeTxnFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { atomic.AddInt32(&f.calls, 1) f.lastReq = req + f.reqs = append(f.reqs, req) if f.resp != nil || f.err != nil { return f.resp, f.err } @@ -304,7 +306,7 @@ func TestRoutedDeleteSpecificVersionUsesWormCondition(t *testing.T) { }, } - if code := s3a.routedDeleteSpecificVersion(owner, "b", "obj", versionId, true, false); code != s3err.ErrAccessDenied { + if code := s3a.routedDeleteSpecificVersion(owner, "b", "obj", versionId, true, false, false); code != s3err.ErrAccessDenied { t.Fatalf("locked version delete returned %v, want %v", code, s3err.ErrAccessDenied) } @@ -343,7 +345,8 @@ func TestRoutedDeleteSpecificVersionUsesWormCondition(t *testing.T) { } recompute := req.Mutations[0] if recompute.Type != filer_pb.ObjectMutation_RECOMPUTE_LATEST || - recompute.GetRecompute().GetExcludeName() != versionFile { + recompute.GetRecompute().GetExcludeName() != versionFile || + recompute.GetRecompute().GetOnlyIfCurrentName() != versionFile { t.Fatalf("recompute mutation = %+v", recompute) } deleteVersion := req.Mutations[1] @@ -356,6 +359,54 @@ func TestRoutedDeleteSpecificVersionUsesWormCondition(t *testing.T) { } } +// TestRoutedDeleteSpecificVersionGuardsRepoint checks that the repoint rides in +// the version's own delete transaction, so the WORM guards gate both, and that +// the recompute carries an only-while-current guard — the filer applies it only +// while the pointer still names the deleted version, so a noncurrent delete +// leaves a cleared (suspended) or moved pointer alone. +func TestRoutedDeleteSpecificVersionGuardsRepoint(t *testing.T) { + versionId := "672a75d526ef29c79fe6b5680cad4a0d" + versionFile := "v_" + versionId + newS3a := func() *S3ApiServer { + return &S3ApiServer{option: &S3ApiServerOption{ + BucketsPath: "/buckets", + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }} + } + + t.Run("one transaction behind the version's lock checks", func(t *testing.T) { + filer := &fakeTxnFiler{} + owner := startFakeFiler(t, filer) + if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, false); code != s3err.ErrNone { + t.Fatalf("delete returned %v, want %v", code, s3err.ErrNone) + } + if len(filer.reqs) != 1 { + t.Fatalf("transactions = %d, want 1", len(filer.reqs)) + } + req := filer.reqs[0] + if req.ConditionKey != "/buckets/b/obj.versions/"+versionFile { + t.Fatalf("ConditionKey = %q", req.ConditionKey) + } + if len(req.Mutations) != 2 || + req.Mutations[0].Type != filer_pb.ObjectMutation_RECOMPUTE_LATEST || + req.Mutations[0].GetRecompute().GetOnlyIfCurrentName() != versionFile || + req.Mutations[1].Type != filer_pb.ObjectMutation_DELETE { + t.Fatalf("mutations = %+v, want guarded recompute + delete", req.Mutations) + } + }) + + t.Run("metadata-only delete keeps the data", func(t *testing.T) { + filer := &fakeTxnFiler{} + owner := startFakeFiler(t, filer) + if code := newS3a().routedDeleteSpecificVersion(owner, "b", "obj", versionId, false, false, true); code != s3err.ErrNone { + t.Fatalf("delete returned %v, want %v", code, s3err.ErrNone) + } + if filer.lastReq.Mutations[1].IsDeleteData { + t.Fatal("metadata-only delete set IsDeleteData") + } + }) +} + func TestWormDeleteConditionForGovernanceBypass(t *testing.T) { cond := wormDeleteCondition(true, true) if len(cond.GetClauses()) != 2 { diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index e37d5d6f8d..4f470a71de 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -178,13 +178,20 @@ func wormDeleteCondition(worm, bypass bool) *filer_pb.WriteCondition { }} } -// routedDeleteSpecificVersion removes one version under the owner filer's object -// lock, first repointing .versions while excluding the deleted version. -func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, bucket, object, versionId string, worm, bypass bool) s3err.ErrorCode { +// routedDeleteSpecificVersion removes one version under the owner filer's +// object lock. The repoint and the delete ride one transaction so the version's +// WORM guards gate both; the recompute is itself guarded to fire only while the +// pointer still names the deleted version — a noncurrent delete must leave it +// alone, since it can point at an older version than a re-derive would pick, or +// at nothing, as when a suspended bucket's null write cleared it while the null +// object stays current. +func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, bucket, object, versionId string, worm, bypass, metadataOnly bool) s3err.ErrorCode { if !isValidVersionID(versionId) { return s3err.ErrInvalidRequest } versionFileName := s3a.getVersionFileName(versionId) + recompute := s3a.latestPointerRecompute(bucket, object, isNewFormatVersionId(versionId), versionFileName, false) + recompute.GetRecompute().OnlyIfCurrentName = versionFileName versionsPath := s3a.toFilerPath(bucket, object+s3_constants.VersionsFolder) cond := wormDeleteCondition(worm, bypass) req := &filer_pb.ObjectTransactionRequest{ @@ -193,8 +200,8 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck ConditionKey: versionsPath + "/" + versionFileName, Condition: cond, Mutations: []*filer_pb.ObjectMutation{ - s3a.latestPointerRecompute(bucket, object, isNewFormatVersionId(versionId), versionFileName, false), - {Type: filer_pb.ObjectMutation_DELETE, Directory: versionsPath, Name: versionFileName, IsDeleteData: true, RemoveEmptyParent: true}, + recompute, + {Type: filer_pb.ObjectMutation_DELETE, Directory: versionsPath, Name: versionFileName, IsDeleteData: !metadataOnly, RemoveEmptyParent: true}, }, } resp, err := s3a.objectTxnOnFiler(owner, req) @@ -219,7 +226,7 @@ func (s3a *S3ApiServer) routedDeleteSpecificVersion(owner pb.ServerAddress, buck // (condition defaults to lock_key). The second return reports whether the delete // was settled here: the raw delete cannot remove an entry other keys are nested // under, which the lock path handles by stripping the object off it instead. -func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass bool) (s3err.ErrorCode, bool) { +func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, object string, worm, bypass, metadataOnly bool) (s3err.ErrorCode, bool) { fullpath := util.NewFullPath(s3a.bucketDir(bucket), object) dir, name := fullpath.DirAndName() resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ @@ -227,7 +234,7 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, RouteKey: s3a.objectRouteKey(bucket, object), Condition: wormDeleteCondition(worm, bypass), Mutations: []*filer_pb.ObjectMutation{ - {Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: true}, + {Type: filer_pb.ObjectMutation_DELETE, Directory: dir, Name: name, IsDeleteData: !metadataOnly}, }, }) switch { diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 2cb5dba999..d2b10c2016 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -321,7 +321,7 @@ func (s3a *S3ApiServer) createNullDeleteMarker(bucket, object string) error { // failures other than not found propagate: a masked read error would answer // a live filer hiccup as a deleted version. func (s3a *S3ApiServer) nullDeleteMarkerEntry(bucket, object string) (*filer_pb.Entry, error) { - entry, err := s3a.getEntry(s3a.getVersionedObjectDir(bucket, s3_constants.NormalizeObjectKey(object)), s3a.getVersionFileName("null")) + entry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, s3_constants.NormalizeObjectKey(object), s3a.getVersionedObjectDir(bucket, s3_constants.NormalizeObjectKey(object)), s3a.getVersionFileName("null")) if err != nil { if errors.Is(err, filer_pb.ErrNotFound) { return nil, nil @@ -1205,13 +1205,12 @@ func (s3a *S3ApiServer) getSpecificObjectVersion(bucket, object, versionId strin if versionId == "" { // Get current version - return s3a.getEntry(s3a.bucketDir(bucket), normalizedObject) + return s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) } if versionId == "null" { // "null" version ID refers to pre-versioning objects stored as regular files - bucketDir := s3a.bucketDir(bucket) - entry, err := s3a.getEntry(bucketDir, normalizedObject) + entry, err := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) if err != nil { return nil, fmt.Errorf("null version object %s not found: %w", normalizedObject, err) } @@ -1221,11 +1220,12 @@ func (s3a *S3ApiServer) getSpecificObjectVersion(bucket, object, versionId strin return entry, nil } - // Get specific version from .versions directory + // Get specific version from .versions directory; it commits on the object's + // write owner, so read it there ahead of replication. versionsDir := s3a.getVersionedObjectDir(bucket, normalizedObject) versionFile := s3a.getVersionFileName(versionId) - entry, err := s3a.getEntry(versionsDir, versionFile) + entry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, versionsDir, versionFile) if err != nil { return nil, fmt.Errorf("version %s not found: %w", versionId, err) } @@ -1845,7 +1845,7 @@ func (s3a *S3ApiServer) clearStaleLatestVersionPointer(bucket, object, bucketDir startFrom := "" for { - entries, isLast, listErr := s3a.list(versionsDir, "", startFrom, false, filer.PaginationSize) + entries, isLast, listErr := s3a.listObjectScopedEntriesRoutedByKey(bucket, object, versionsDir, "", startFrom, false, filer.PaginationSize) if listErr != nil { glog.Warningf("%s: re-scan failed for %s/%s, leaving pointer untouched: %v", caller, bucket, object, listErr) return false @@ -1860,7 +1860,7 @@ func (s3a *S3ApiServer) clearStaleLatestVersionPointer(bucket, object, bucketDir startFrom = entries[len(entries)-1].Name } - liveEntry, err := s3a.getEntry(bucketDir, versionsObjectPath) + liveEntry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, object, bucketDir, versionsObjectPath) if err != nil { // Directory was concurrently removed - reader will get NoSuchKey // via the clean-miss path; pointer is effectively cleared. @@ -1891,7 +1891,7 @@ func (s3a *S3ApiServer) clearStaleLatestVersionPointer(bucket, object, bucketDir delete(updated.Extended, s3_constants.ExtLatestVersionFileNameKey) clearCachedVersionMetadata(updated.Extended) updated.Name = versionsObjectPath - if err := s3a.conditionalUpdateEntry(bucketDir, updated, expected); err != nil { + if err := s3a.conditionalUpdateEntryRouted(bucket, object, bucketDir, updated, expected); err != nil { switch status.Code(err) { case codes.FailedPrecondition: // A writer moved the pointer after this heal's snapshot; its @@ -1987,9 +1987,11 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri glog.V(1).Infof("doGetLatestObjectVersion: looking for latest version of %s/%s (normalized: %s, retries: %d)", bucket, object, normalizedObject, maxRetries) - // Get the .versions directory entry to read latest version metadata with retry logic for filer consistency + // Get the .versions directory entry to read latest version metadata with retry logic for filer consistency. + // The pointer and the version files commit on the object's write owner, so + // read them there ahead of replication. versionsEntry, err := lookupVersionsEntryWithRetry(func() (*filer_pb.Entry, error) { - return s3a.getEntry(bucketDir, versionsObjectPath) + return s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsObjectPath) }, maxRetries) if err != nil { @@ -2003,7 +2005,7 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri // regular (non-versioned) object file. glog.V(1).Infof("getLatestObjectVersion: no .versions directory for %s/%s after %d attempts (error: %v), checking for pre-versioning object", bucket, normalizedObject, maxRetries, err) - regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) if regularErr != nil { if !isObjectEntryMiss(regularErr) { return nil, fmt.Errorf("failed to get regular object %s/%s: %w", bucket, normalizedObject, regularErr) @@ -2026,7 +2028,7 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri for metaAttempt := 1; metaAttempt <= metadataRetries; metaAttempt++ { // Small delay and re-read the directory time.Sleep(time.Millisecond * 100) - versionsEntry, err = s3a.getEntry(bucketDir, versionsObjectPath) + versionsEntry, err = s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsObjectPath) if err != nil { break } @@ -2061,7 +2063,7 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri // Get the actual latest version file entry latestVersionPath := versionsObjectPath + "/" + latestVersionFile - latestVersionEntry, err := s3a.getEntry(bucketDir, latestVersionPath) + latestVersionEntry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, latestVersionPath) if err != nil { // The pointer refers to a version file that no longer exists. Rather than // surfacing a hard error that requires manual repair, rescan the .versions @@ -2085,9 +2087,7 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri // regular path is current, so that object wins; only when it is absent do we // rescan .versions/ to rebuild a pointer lost while real version files remain. func (s3a *S3ApiServer) recoverLatestVersionWithoutPointer(bucket, normalizedObject string, versionsEntry *filer_pb.Entry) (*filer_pb.Entry, error) { - bucketDir := s3a.bucketDir(bucket) - - if regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject); regularErr == nil && isNullVersionObject(regularEntry, normalizedObject) { + if regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject); regularErr == nil && isNullVersionObject(regularEntry, normalizedObject) { return regularEntry, nil } @@ -2163,6 +2163,14 @@ func (s3a *S3ApiServer) scanLatestVersionEntry(versionsDir string) (latestEntry return scanLatestVersionEntry(s3a.list, versionsDir) } +// objectScopedLister binds listObjectScopedEntriesRoutedByKey to an object so a +// .versions scan reads where the object's writes commit, ahead of replication. +func (s3a *S3ApiServer) objectScopedLister(bucket, object string) entryLister { + return func(parentDirectoryPath, prefix, startFrom string, inclusive bool, limit uint32) ([]*filer_pb.Entry, bool, error) { + return s3a.listObjectScopedEntriesRoutedByKey(bucket, object, parentDirectoryPath, prefix, startFrom, inclusive, limit) + } +} + func scanLatestVersionEntry(list entryLister, versionsDir string) (latestEntry *filer_pb.Entry, latestVersionId, latestVersionFileName string, isDeleteMarker bool, err error) { startFrom := "" for { @@ -2213,7 +2221,7 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s // and the caller renders NoSuchKey (with x-amz-delete-marker) from the // returned entry. Restricting to content versions here would "undelete" // the object by promoting an older content version over a newer marker. - latestEntry, latestVersionId, latestVersionFileName, isDeleteMarker, scanErr := s3a.scanLatestVersionEntry(versionsDir) + latestEntry, latestVersionId, latestVersionFileName, isDeleteMarker, scanErr := scanLatestVersionEntry(s3a.objectScopedLister(bucket, normalizedObject), versionsDir) if scanErr != nil { return nil, scanErr } @@ -2250,7 +2258,7 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s // still returns the scanned entry, and the winner's pointer needs no // repair. Write the live Extended map so fields a concurrent writer // updated between the re-fetch and the persist are preserved. - liveEntry, liveErr := s3a.getEntry(bucketDir, versionsObjectPath) + liveEntry, liveErr := s3a.getObjectScopedEntryRoutedByKey(bucket, normalizedObject, bucketDir, versionsObjectPath) if liveErr != nil { // The directory is gone (e.g. a concurrent delete's teardown); nothing // to repair, and this read still returns the scanned entry. @@ -2282,7 +2290,7 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s updated.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(latestVersionFileName) setCachedListMetadata(updated, latestEntry) - updateErr := s3a.conditionalUpdateEntry(bucketDir, updated, expected) + updateErr := s3a.conditionalUpdateEntryRouted(bucket, normalizedObject, bucketDir, updated, expected) switch { case updateErr == nil: versioningHealInfof("healed", "bucket=%s key=%s mode=pointer_repaired new_version=%s file=%s delete_marker=%v", bucket, normalizedObject, latestVersionId, latestVersionFileName, isDeleteMarker) @@ -2299,6 +2307,62 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s return latestEntry, nil } +// conditionalUpdateEntryRouted persists the repair on the object's write +// owner — the filer the heal read from — via an ObjectTransaction carrying +// the object's route key. A conditioned UpdateEntry would instead route by +// the .versions path's own key, which can hash to a different filer, leaving +// the pointer on the owner the reads consult unchanged. The IF_ENTRY_EQUAL +// clause keeps the CAS semantics; a precondition failure or a missing entry +// both surface as FailedPrecondition, matching what the callers map. +// The lock goes on the object path — the lock version writes and deletes +// serialize on — while condition_key keeps the check on the .versions entry, +// so the repair cannot pass its precondition and then apply over a pointer a +// writer advanced in between. +func (s3a *S3ApiServer) conditionalUpdateEntryRouted(bucket, object, directory string, entry, expected *filer_pb.Entry) error { + owner := s3a.routableWriteOwner(bucket, object) + if owner == "" { + return s3a.conditionalUpdateEntry(directory, entry, expected) + } + setExtended := make(map[string][]byte) + for k, v := range entry.Extended { + if ev, ok := expected.Extended[k]; !ok || !bytes.Equal(ev, v) { + setExtended[k] = v + } + } + var deleteExtended []string + for k := range expected.Extended { + if _, ok := entry.Extended[k]; !ok { + deleteExtended = append(deleteExtended, k) + } + } + resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ + LockKey: s3a.toFilerPath(bucket, object), + RouteKey: s3a.objectRouteKey(bucket, object), + ConditionKey: string(util.NewFullPath(directory, entry.Name)), + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ + Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, + ExpectedEntry: expected, + }}}, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PATCH_EXTENDED, + Directory: directory, + Name: entry.Name, + SetExtended: setExtended, + DeleteExtended: deleteExtended, + }}, + }) + if err != nil { + return err + } + if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { + return status.Error(codes.FailedPrecondition, resp.Error) + } + if resp.Error != "" { + return errors.New(resp.Error) + } + return nil +} + // conditionalUpdateEntry persists entry only while the stored image still // equals expected: the filer evaluates IF_ENTRY_EQUAL under the entry's path // lock, and conditional writes route to the owner filer, so a writer @@ -2499,12 +2563,12 @@ func (s3a *S3ApiServer) recoverLatestListEntryByScan(bucket, normalizedObject st // object current (the write clears the pointer and leaves the explicit // nullIsLatest signal), or that the pointer has not replicated to this // filer while the version files have. - regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) + regularEntry, regularErr := s3a.getObjectEntryRoutedByKey(bucket, normalizedObject) if regularErr == nil && nullIsLatest { return regularEntry, nil } - latestEntry, latestVersionId, _, isDeleteMarker, err := s3a.scanLatestVersionEntry(versionsDir) + latestEntry, latestVersionId, _, isDeleteMarker, err := scanLatestVersionEntry(s3a.objectScopedLister(bucket, normalizedObject), versionsDir) if err != nil { return nil, err } diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go index fab96e187e..689ea5e59f 100644 --- a/weed/s3api/s3api_remote_storage_test.go +++ b/weed/s3api/s3api_remote_storage_test.go @@ -590,8 +590,9 @@ func TestCacheRemoteObjectForStreamingCached(t *testing.T) { cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { return &filer_pb.CacheRemoteObjectToLocalClusterResponse{ Entry: &filer_pb.Entry{ - Name: req.Name, - Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}}, + Name: req.Name, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}}, }, }, nil }, @@ -605,6 +606,57 @@ func TestCacheRemoteObjectForStreamingCached(t *testing.T) { assert.Len(t, got.GetChunks(), 1) } +// TestCacheRemoteObjectForCopyStaleGeneration pins that a cache response for a +// different generation of the key is rejected rather than swapped in as the +// authorized copy source. +func TestCacheRemoteObjectForCopyStaleGeneration(t *testing.T) { + filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{ + cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{ + Entry: &filer_pb.Entry{ + Name: req.Name, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteETag: "stale"}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}}, + }, + }, nil + }, + }) + s3a := newRemoteCacheTestServer(filerAddr) + + got := s3a.cacheRemoteObjectForCopy(context.Background(), + &filer_pb.Entry{Name: "obj.bin", RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteETag: "fresh"}}, + "mybucket", "obj.bin", "") + assert.Nil(t, got, "a different generation must not replace the resolved source") +} + +// TestS3VersionedReadRejectsStaleCacheFill pins that a versioned GET does not +// stream a cached entry from a different generation of the object. +func TestS3VersionedReadRejectsStaleCacheFill(t *testing.T) { + filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{ + cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{ + Entry: &filer_pb.Entry{ + Name: req.Name, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteETag: "stale"}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}}, + }, + }, nil + }, + }) + s3a := newRemoteCacheTestServer(filerAddr) + + entry := &filer_pb.Entry{ + Name: "obj.bin", + Attributes: &filer_pb.FuseAttributes{FileSize: 100}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteETag: "fresh"}, + } + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/obj.bin", nil) + err := s3a.streamFromVolumeServers(w, r, entry, "", "mybucket", "obj.bin", "v123") + require.Error(t, err) + assert.Equal(t, http.StatusServiceUnavailable, w.Code, w.Body.String()) +} + // fakeStreamRemoteClient records the ReadFileAsStream call and serves from an // in-memory byte slice. The embedded nil interface panics on any other call. type fakeStreamRemoteClient struct { diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index dc9cb139a2..36ccad3768 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -364,7 +364,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl } s3ApiServer.objectWriteLockClient = objectWriteLockClient s3ApiServer.newObjectWriteLock = func(bucket, object string) (objectWriteLock, error) { - lockKey := objectWriteRouteKeyPrefix + s3ApiServer.toFilerPath(bucket, object) + lockKey := s3ApiServer.objectRouteKey(bucket, object) owner := fmt.Sprintf("s3api-%d", s3ApiServer.randomClientId) lock := objectWriteLockClient.NewShortLivedLock(lockKey, owner) if lock == nil { diff --git a/weed/s3api/s3api_version_id.go b/weed/s3api/s3api_version_id.go index 124db3cf07..c481f54f7f 100644 --- a/weed/s3api/s3api_version_id.go +++ b/weed/s3api/s3api_version_id.go @@ -166,7 +166,7 @@ func (s3a *S3ApiServer) getVersionIdFormat(bucket, object string) bool { versionsPath := object + s3_constants.VersionsFolder // Try to get the .versions directory entry - versionsEntry, err := s3a.getEntry(bucketDir, versionsPath) + versionsEntry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, object, bucketDir, versionsPath) if err != nil { // .versions directory doesn't exist yet - use new format return true diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index c6295f7810..21f823410b 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -292,23 +292,30 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr } else { glog.V(3).InfofCtx(ctx, "CreateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), createErr) resp.Error = createErr.Error() - switch { - case errors.Is(createErr, filer_pb.ErrEntryNameTooLong): - resp.ErrorCode = filer_pb.FilerError_ENTRY_NAME_TOO_LONG - case errors.Is(createErr, filer_pb.ErrParentIsFile): - resp.ErrorCode = filer_pb.FilerError_PARENT_IS_FILE - case errors.Is(createErr, filer_pb.ErrExistingIsDirectory): - resp.ErrorCode = filer_pb.FilerError_EXISTING_IS_DIRECTORY - case errors.Is(createErr, filer_pb.ErrExistingIsFile): - resp.ErrorCode = filer_pb.FilerError_EXISTING_IS_FILE - case errors.Is(createErr, filer_pb.ErrEntryAlreadyExists): - resp.ErrorCode = filer_pb.FilerError_ENTRY_ALREADY_EXISTS - } + resp.ErrorCode = filerErrorCode(createErr) } return } +// filerErrorCode maps a filer error to its proto code so a client can match it +// with errors.Is against the sentinels instead of parsing the message. +func filerErrorCode(err error) filer_pb.FilerError { + switch { + case errors.Is(err, filer_pb.ErrEntryNameTooLong): + return filer_pb.FilerError_ENTRY_NAME_TOO_LONG + case errors.Is(err, filer_pb.ErrParentIsFile): + return filer_pb.FilerError_PARENT_IS_FILE + case errors.Is(err, filer_pb.ErrExistingIsDirectory): + return filer_pb.FilerError_EXISTING_IS_DIRECTORY + case errors.Is(err, filer_pb.ErrExistingIsFile): + return filer_pb.FilerError_EXISTING_IS_FILE + case errors.Is(err, filer_pb.ErrEntryAlreadyExists): + return filer_pb.FilerError_ENTRY_ALREADY_EXISTS + } + return filer_pb.FilerError_OK +} + // ObjectTransaction applies an ordered list of entry mutations atomically with // respect to other writers of the same object, by holding the per-path lock on // lock_key for the whole call. The optional condition is checked first, against @@ -434,6 +441,7 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje 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) return resp, nil } } @@ -707,6 +715,10 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj return err } + if rc.OnlyIfCurrentName != "" && string(pointer.Extended[rc.NameToKey]) != rc.OnlyIfCurrentName { + return nil + } + // Capture the pre-update image so the metadata notification carries a correct // diff; pointer.Extended is mutated in place below. oldPointer := pointer.ShallowClone() diff --git a/weed/server/filer_grpc_server_recompute_test.go b/weed/server/filer_grpc_server_recompute_test.go index 18ecc6848f..c22fa2efed 100644 --- a/weed/server/filer_grpc_server_recompute_test.go +++ b/weed/server/filer_grpc_server_recompute_test.go @@ -138,3 +138,68 @@ func TestRecomputeLatestDemoteEmitsEvent(t *testing.T) { t.Fatalf("missing demote event: %+v", queue.snapshot()) } } + +// OnlyIfCurrentName scopes a recompute to the case where the pointer still +// names a given child — a version delete repoints only while the pointer names +// the deleted version, so a noncurrent delete cannot resurrect an older +// version over a cleared or moved pointer. +func TestRecomputeLatestOnlyIfCurrentName(t *testing.T) { + newMutation := func() *filer_pb.ObjectMutation { + return &filer_pb.ObjectMutation{ + Type: filer_pb.ObjectMutation_RECOMPUTE_LATEST, + Directory: "/buckets/b", + Name: "obj.versions", + Recompute: &filer_pb.Recompute{ + ScanDir: "/buckets/b/obj.versions", + Descending: true, + NameToKey: "latest-file", + CopyExtended: map[string]string{"latest-vid": "vid"}, + ExcludeName: "v_200", + OnlyIfCurrentName: "v_200", + }, + } + } + newStore := func(latestFile string) *renameTestStore { + store := newRenameTestStore() + pointer := newDirectoryEntry("/buckets/b/obj.versions", 10) + pointer.Extended = map[string][]byte{"latest-file": []byte(latestFile)} + store.entries["/buckets/b/obj.versions"] = pointer + v100 := newFileEntry("/buckets/b/obj.versions/v_100", 11) + v100.Extended = map[string][]byte{"vid": []byte("100")} + store.entries["/buckets/b/obj.versions/v_100"] = v100 + return store + } + + t.Run("pointer names the deleted version", func(t *testing.T) { + store := newStore("v_200") + server := &FilerServer{filer: newRenameTestFiler(t, store)} + if err := server.applyObjectMutation(context.Background(), newMutation(), false, nil); err != nil { + t.Fatalf("applyObjectMutation: %v", err) + } + ptr, err := store.FindEntry(context.Background(), "/buckets/b/obj.versions") + if err != nil { + t.Fatalf("find pointer: %v", err) + } + if got := string(ptr.Extended["latest-file"]); got != "v_100" { + t.Fatalf("pointer latest-file = %q, want repointed to v_100", got) + } + }) + + t.Run("pointer names another version", func(t *testing.T) { + store := newStore("v_other") + server := &FilerServer{filer: newRenameTestFiler(t, store)} + if err := server.applyObjectMutation(context.Background(), newMutation(), false, nil); err != nil { + t.Fatalf("applyObjectMutation: %v", err) + } + ptr, err := store.FindEntry(context.Background(), "/buckets/b/obj.versions") + if err != nil { + t.Fatalf("find pointer: %v", err) + } + if got := string(ptr.Extended["latest-file"]); got != "v_other" { + t.Fatalf("pointer latest-file = %q, want untouched v_other", got) + } + if _, copied := ptr.Extended["latest-vid"]; copied { + t.Fatalf("pointer latest-vid copied on a skipped recompute: %+v", ptr.Extended) + } + }) +}