s3: route regular copy source reads to the write owner (#12173)

* 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>

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
 	}