From 74f196ceab3d427e328d783cc567b0b1c6b1d5ba Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 28 Oct 2026 14:35:39 -0700 Subject: [PATCH] s3: route regular copy source reads to the write owner (#12173) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * s3: read regular copy sources from their write owner * filer: report mutation sentinel codes in ObjectTransaction responses A failed mutation only surfaced resp.Error text, so routed writers could not distinguish an existing-directory conflict from a generic failure and could not apply the prefix-object retry the plain CreateEntry path has. * s3api: retry a routed mkfile as a prefix object over a directory mkFile retries ErrExistingIsDirectory by marking the entry a prefix object; the routed path returned the conflict instead, so a routed write to a key that is also a prefix failed where the lock path succeeds. The filer now reports the code on the transaction response, making the retry recognizable. * s3api: commit copy destinations on the key's write owner An owner-routed source read still left the destination on the gateway's default filer, so a second copy resolved the first copy's destination on the owner, found nothing, and returned NoSuchKey. Write unversioned and suspended destinations through routedMkFile, and write the version file on the owner the finalize recomputes on. * s3api: cache remote objects on the filer that served the entry An owner-routed lookup can return a remote-only entry that has not replicated to the gateway's default filer; the cache RPC then cannot find it and the copy or read degrades to a 503. Ask the entry's write owner to cache it first, falling back to the default filer. * s3: read versioned object state from the key's write owner CopyObject destinations now commit version files and the .versions pointer on the owner's object-scope ring key, so versioned reads through the default filer could miss them until replication. Generalize the routed read to dir/name lookups and paginated lists scoped to the object's owner, and point the versioned read paths at it: getSpecificObjectVersion, doGetLatestObjectVersion, the .versions quick-checks in Get/Head/attributes, the self-heal rescans, tagging, delete conditionals, directory-marker history checks, rename destination, multipart completion checks, and the version-id format probe. Reads fall back to the default filer when no ring owner is resolvable, preserving single-filer deployments. * test(s3api): fake RECOMPUTE_LATEST/DELETE and cover versioned copy reads The ambiguous-PUT fake filer now creates parent directories on PUT (mirroring InsertEntry), applies DELETE mutations, and re-derives the .versions pointer for RECOMPUTE_LATEST the way applyRecomputeLatest does, so owner-vs-replica tests can exercise versioned writes. TestVersionedCopyDestinationReadsOnOwner copies into a versioned bucket whose writes commit on the owner while the gateway's default filer is a replica with no replication, then asserts reads by versionId, latest-version resolution, and a copy chained off the new version all resolve on the owner. * test(s3api): gofmt the versioned-copy coverage * s3: only swap in a cache fill of the resolved remote generation A cache response can come back from a filer holding an older generation of the key. Replacing the authorized source or swapping its chunks under a size and range computed for the resolved entry serves the wrong object or truncates the response. Match RemoteEntry identity and size before using a cached entry, in both the copy and streaming paths. * filer: keep directory attributes when an object lands as prefix object The object-onto-directory retry replaced the existing directory's mode, owner, and extended attributes with the object's. Preserve the directory identity; merge directory-held extended keys only when the old entry never held an object, so a prior generation's object metadata is not carried forward. * s3: keep marker lock checks on the filer that replaces the marker The object-lock precondition read the key's write owner while mkdir still replaces the marker on the default filer, so a locked marker not yet replicated to the owner could be overwritten. Read the entry where the replacement lands, in both PutObjectHandler and zip extraction. Also hash a trailing-slash marker key to the same route key as the object it names, so marker and object share one owner and one write lock. * s3: route tag writes to the owner tag reads resolve GetObjectTagging reads the key's write owner, but PutObjectTagging and DeleteObjectTagging still wrote through the default filer, so a successful change returned the owner's stale tags until replication. setTags, rmTags, and the versioned tag updates now write on the owner first, falling back to the default filer when no owner resolves. * s3: keep rename lookups on the filer that performs the move AtomicRenameEntry and the retry token it carries both execute on the default filer, but the source and destination reads went to the write owner — a completed rename's token was invisible to its own retry, and a source that only exists on the default filer looked absent. Read both ends on the default filer. * s3: send the .versions pointer repair with the object's route key The heal reads the .versions entry on the object's owner, but the conditional UpdateEntry is routed by the .versions path's own ring key, which can hash to a different filer — leaving the pointer the routed reads consult unchanged. Ship the CAS repair as a PATCH_EXTENDED ObjectTransaction on the object's route key instead. * filer: preserve only directory identity on a prefix-object write Merging the old directory's extended attributes could carry a marker's encryption, version, or etag fields onto the new plaintext object, since a directory's own keys cannot be told apart from the object it held. Keep only mode, uid, gid, user name and group names. * s3: also require matching size when a cache fill replaces the source The remote-generation check covers identity; size equality is the cheap guard that keeps the streamed or copied range consistent with what was resolved. * s3: lock the object path for a routed .versions pointer repair conditionalUpdateEntryRouted held the lock on the .versions path while version writes and deletes serialize on the object path, so a repair could pass IF_ENTRY_EQUAL and then apply over a pointer a writer had advanced in between. Lock the object path like the writers and keep the condition on the .versions entry via condition_key. * s3: do not replay an owner-side tag mutation on another filer withWriteOwnerFilerClient re-ran fn on the default filer after any owner error, so an authoritative not-found could update a stale replica entry and still report success, and a post-apply transport error could apply the mutation twice. The failover walk now ends on any error fn returns; only a channel that never came up moves to another filer. * filer: keep a directory's xattrs when a prefix object replaces it The prefix-object branch preserves the directory's identity but dropped its Extended map wholesale; FUSE xattrs live there under the xattr- prefix and belong to the directory, not to the object the marker held. * s3: delete null and specific versions on the filer that serves them On the default filer an absent entry deletes as success, so a version that only reached the write owner survived a reported success while owner-routed reads kept serving it. The suspended-versioning and lifecycle deletes now go through routedDeleteNullVersion and routedDeleteSpecificVersion when an owner resolves; an unsettled attempt keeps the lock path. * s3: route a lifecycle null-version delete to the object entry The "null" version is the regular object entry, not a .versions file, so it goes through routedDeleteNullVersion; only a real version id takes the .versions-file delete. * s3: repoint .versions only while it still names the deleted version routedDeleteSpecificVersion recomputed the latest pointer on every version delete, so removing a noncurrent version re-derived the pointer to the newest remaining file — resurrecting an older version over a cleared pointer (a suspended bucket's null write keeps it clear while the null object stays current) or over a pointer that moved on. The pointer is now read on the owner and repointed under an IF_ENTRY_EQUAL guard only while it still names the deleted version; a contested pointer is left for the self-heal path rather than overwritten by a stale scan. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: let lifecycle expiry keep its metadata-only deletes on routed paths The routed lifecycle deletes hardcoded IsDeleteData, so expiring an entry with a volume TTL queued per-chunk deletions instead of leaving the data for the volume to reclaim — extra work on the owner for deletes recordMetadataOnlyIf still counts as metadata-only. Pass the caller's cleanup choice through routedDeleteNullVersion and routedDeleteSpecificVersion; the interactive DELETE path keeps deleting data. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: let RECOMPUTE_LATEST skip a pointer that moved on A version delete repoints .versions only while the pointer still names the deleted version, but the gateway cannot carry the pointer's current state into the transaction's condition — the condition slot is already spent on the version's WORM guards. only_if_current_name moves the check under the transaction lock: the filer compares the pointer's name_to_key value before scanning and skips the recompute when it moved (or was cleared by a suspended bucket's null write), so a noncurrent delete cannot resurrect an older version. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: repoint inside the version's locked delete repointLatestBeforeVersionDelete ran in its own transaction ahead of the delete, so a version delete denied by retention or legal hold still moved the latest pointer — GET and HEAD went on serving the older version it rolled to. The repoint and the delete now ride one transaction behind the version's WORM guards, atomically gated by the new only_if_current_name recompute guard instead of a stale-read IF_ENTRY_EQUAL. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- other/java/client/src/main/proto/filer.proto | 1 + weed/filer/filer.go | 19 ++ weed/filer/filer_prefix_object_test.go | 45 +++- weed/pb/filer.proto | 1 + weed/pb/filer_pb/filer.pb.go | 38 ++-- weed/pb/filer_pb/filer_vtproto.pb.go | 43 ++++ weed/s3api/filer_multipart.go | 9 +- weed/s3api/filer_util_tags.go | 9 +- weed/s3api/s3api_copy_source_classify_test.go | 199 ++++++++++++++++++ weed/s3api/s3api_directory_marker.go | 11 +- weed/s3api/s3api_internal_lifecycle.go | 72 ++++++- weed/s3api/s3api_object_handlers.go | 92 ++++++-- .../s3api/s3api_object_handlers_attributes.go | 4 +- weed/s3api/s3api_object_handlers_copy.go | 98 +++++---- weed/s3api/s3api_object_handlers_delete.go | 34 ++- weed/s3api/s3api_object_handlers_extract.go | 3 + weed/s3api/s3api_object_handlers_put.go | 3 + ...3api_object_handlers_put_ambiguous_test.go | 102 ++++++++- weed/s3api/s3api_object_handlers_rename.go | 12 +- weed/s3api/s3api_object_handlers_tagging.go | 17 +- weed/s3api/s3api_object_routed_read.go | 44 +++- weed/s3api/s3api_object_routed_write.go | 34 ++- weed/s3api/s3api_object_routed_write_test.go | 55 ++++- weed/s3api/s3api_object_versioned_finalize.go | 21 +- weed/s3api/s3api_object_versioning.go | 108 ++++++++-- weed/s3api/s3api_remote_storage_test.go | 56 ++++- weed/s3api/s3api_server.go | 2 +- weed/s3api/s3api_version_id.go | 2 +- weed/server/filer_grpc_server.go | 36 ++-- .../filer_grpc_server_recompute_test.go | 65 ++++++ 30 files changed, 1065 insertions(+), 170 deletions(-) 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) + } + }) +}