fix(s3/versioning): repair dangling latest-version pointer after partial delete (#9460)

* fix(s3/versioning): repair dangling latest-version pointer after partial delete

deleteSpecificObjectVersion did two non-atomic filer ops: rm the version
blob, then update the .versions/ pointer. Step 2 failures were silently
logged and the client got 204 OK, so any transient blip (filer timeout,
process restart between RPCs, lock contention) left the .versions/
directory naming a missing file. Subsequent GETs paid the 10-retry
self-heal cost and returned NoSuchKey — surfacing as "Storage not found"
to Veeam, which is what triggered this investigation.

Three changes:

1. Pre-roll the pointer for the singleton / multi-version-deleting-latest
   cases. The pointer is repointed (multi) or cleared (singleton) before
   the blob rm. A failure between leaves a recoverable orphan blob —
   pointer is consistent, GETs succeed or correctly miss without
   entering the stale-pointer self-heal path.

2. Wrap the load-bearing filer ops in updateLatestVersionAfterDeletion
   with bounded retries (~6.3s worst case). When retries are exhausted
   the function now returns a non-nil error instead of swallowing it.
   The caller logs at Error level and queues the path for the
   reconciler.

3. Background reconciler drains stranded .versions/ pointer-to-missing
   states off the hot path. Bounded in-memory queue with capped retries;
   read-path heal remains as a last-resort safety net.

* fix(s3/versioning): address review on #9460

Four fixes addressing review on PR #9460. All four are correctness;
no behavioural change for the happy path.

1. repointLatestBeforeDeletion: discriminate NotFound from transient
   errors when re-fetching the .versions/ entry. Previously any error
   returned rolled=true,nil — a transient filer hiccup at that point
   would cause the caller to skip the post-delete reconciliation AND
   proceed with the blob rm, producing exactly the dangling pointer
   state the PR aims to prevent. NotFound stays "vacuously consistent"
   (directory already gone); other errors surface so the caller aborts
   before removing the blob.

2. Move the singleton .versions/ teardown out of
   repointLatestBeforeDeletion (where it ran BEFORE the blob rm and
   always failed with "non-empty folder") into deleteSpecificObjectVersion
   AFTER the blob rm. Adds a wasSingleton return value so the caller
   knows when to run the teardown. Without this, every singleton-version
   delete in a versioned bucket leaked an empty .versions/ directory.

3. Wrap the list, getEntry, and mkFile calls inside
   repointLatestBeforeDeletion with retryFilerOp so the pre-roll has
   the same transient-failure resilience as the post-roll path. Without
   retries, a single transient blip causes the caller to fall back to
   the legacy non-atomic flow even when the filer recovers immediately.

4. healVersionsPointer in the reconciler: same NotFound-vs-transient
   discrimination on both the .versions/ getEntry and the latest-file
   presence probe. Previously a transient filer error would silently
   evict the candidate from the queue as "healed", leaving the real
   stranded state until a client read happened to surface it.

Also fixes the gemini-flagged consistency nit: the queued-for-reconciler
error log now uses normalizedObject instead of object so it matches the
queue entry's key.

* fix(s3/versioning): short-circuit terminal errors in retryFilerOp

Add isRetryableFilerErr that returns false for filer_pb.ErrNotFound,
gRPC NotFound, context.Canceled, and context.DeadlineExceeded.
retryFilerOp now bails immediately on a terminal error and returns it
unwrapped, so callers like repointLatestBeforeDeletion.getEntry and
updateLatestVersionAfterDeletion.rm see the raw NotFound instead of
paying the ~6.3 s retry-budget delay AND parsing it out of an
"exhausted N retries" wrapper.

errors.Is and status.Code already walk the %w chain so today's call
sites still work, but the delay was real on the hot DELETE path
whenever a key was genuinely absent. Test added covering all five
terminal-error shapes — each must run the wrapped fn exactly once and
return in under 50 ms.

weed/s3api/s3api_object_versioning.go

@@ -5,6 +5,7 @@ package s3api
 
 import (
 	"bytes"
+	"context"
 	"encoding/hex"
 	"encoding/xml"
 	"errors"
@@ -1044,6 +1045,28 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(bucket, object, versionId st
 		}
 	}
 
+	// Option 1: when deleting the current latest, repoint the .versions/
+	// pointer BEFORE removing the blob. A failure between the two steps
+	// then leaves a recoverable orphan blob (the pointer is already
+	// consistent, GETs serve the prior version or NoSuchKey correctly)
+	// instead of a dangling pointer (which forces every subsequent GET
+	// through the 10-retry self-heal path and returns NoSuchKey for
+	// objects whose latest version was singleton).
+	var (
+		prePointerRolled bool
+		preWasSingleton  bool
+	)
+	if isLatestVersion {
+		rolled, singleton, prepErr := s3a.repointLatestBeforeDeletion(bucket, normalizedObject, versionId)
+		if prepErr != nil {
+			// Surface to the client so they can retry the DELETE. The
+			// blob has NOT been removed yet, so retrying is safe.
+			return fmt.Errorf("failed to repoint latest version before deleting %s: %w", versionId, prepErr)
+		}
+		prePointerRolled = rolled
+		preWasSingleton = singleton
+	}
+
 	// Attempt to delete the version file
 	// Note: We don't check if the file exists first to avoid race conditions
 	// The deletion operation should be idempotent
@@ -1059,24 +1082,254 @@ func (s3a *S3ApiServer) deleteSpecificObjectVersion(bucket, object, versionId st
 		return fmt.Errorf("failed to delete version %s: %v", versionId, deleteErr)
 	}
 
-	// If we deleted the latest version, update the .versions directory metadata to point to the new latest
-	if isLatestVersion {
-		err := s3a.updateLatestVersionAfterDeletion(bucket, normalizedObject)
-		if err != nil {
-			glog.Warningf("deleteSpecificObjectVersion: failed to update latest version after deletion: %v", err)
-			// Don't return error since the deletion was successful
+	switch {
+	case isLatestVersion && prePointerRolled && preWasSingleton:
+		// Pre-roll cleared a singleton pointer. The blob is now gone,
+		// so the .versions/ directory should be empty — try to tear it
+		// down. Non-recursive: any orphan from older code paths leaves
+		// the directory in place for the empty-folder cleaner or our
+		// reconciler to handle.
+		if rmErr := s3a.rm(s3a.bucketDir(bucket), normalizedObject+s3_constants.VersionsFolder, true, false); rmErr != nil {
+			glog.V(2).Infof("deleteSpecificObjectVersion: deferring .versions/ teardown for %s/%s: %v", bucket, normalizedObject, rmErr)
+		}
+	case isLatestVersion && !prePointerRolled:
+		// Multi-version case where another version still exists, or the
+		// pre-step decided the pointer was no longer ours to roll. Run
+		// the post-deletion reconciliation: it updates the pointer to
+		// the new latest and tears down .versions/ when nothing remains.
+		if err := s3a.updateLatestVersionAfterDeletion(bucket, normalizedObject); err != nil {
+			// Option 2: surface this so the operator sees the load-bearing
+			// failure, and queue the path for the reconciler to retry off
+			// the hot path. The blob delete already succeeded, so we don't
+			// want to fail the client request (Veeam et al. treat 5xx on
+			// DELETE as a hard storage error) — but we MUST drive the
+			// pointer to consistency, otherwise the next read pays the
+			// self-heal cost.
+			glog.Errorf("deleteSpecificObjectVersion: failed to update latest version after deletion for %s/%s (queued for reconciler): %v", bucket, normalizedObject, err)
+			if s3a.versionsHealQueue != nil {
+				s3a.versionsHealQueue.Enqueue(bucket, normalizedObject)
+			}
 		}
 	}
 
 	return nil
 }
 
+// repointLatestBeforeDeletion is the "option 1" pre-step for deleting a
+// version that the .versions/ pointer currently names. It scans the
+// directory to find the next-newest version excluding the one about to
+// be deleted, and either repoints to it (multi-version case) or clears
+// the pointer entirely (single-version case). When this returns rolled
+// = true the caller should NOT run updateLatestVersionAfterDeletion
+// after the blob rm — the pointer is already consistent.
+//
+// wasSingleton = true indicates the caller cleared a singleton pointer.
+// After the caller's blob rm completes, deleteSpecificObjectVersion is
+// expected to run the post-deletion .versions/ teardown so the now-empty
+// directory entry is removed; doing that teardown inside this function
+// (i.e. before the blob rm) always fails with "non-empty folder" because
+// the blob is still present.
+//
+// When this returns rolled = false the .versions/ pointer was not in
+// sync with the caller's view (some concurrent writer changed it) and
+// the current deletion is no longer touching the latest; the caller
+// proceeds with the historical multi-step path which will re-snapshot
+// the state inside updateLatestVersionAfterDeletion.
+//
+// All load-bearing filer ops here are wrapped in retryFilerOp so the
+// pre-roll matches the resilience of the post-roll path; without this,
+// a transient filer hiccup during the pre-step would cause the caller
+// to fall back to the legacy non-atomic flow.
+func (s3a *S3ApiServer) repointLatestBeforeDeletion(bucket, normalizedObject, versionIdToDelete string) (rolled bool, wasSingleton bool, err error) {
+	bucketDir := s3a.bucketDir(bucket)
+	versionsObjectPath := normalizedObject + s3_constants.VersionsFolder
+	versionsDir := bucketDir + "/" + versionsObjectPath
+
+	// Find the chronologically newest *other* version (excluding the
+	// one we're about to delete). For singleton objects this returns
+	// nil; for multi-version it returns the previous version.
+	var (
+		newLatestEntry        *filer_pb.Entry
+		newLatestVersionId    string
+		newLatestVersionFile  string
+		newLatestIsDeleteMark bool
+		startFrom             string
+	)
+	for {
+		var (
+			entries []*filer_pb.Entry
+			isLast  bool
+		)
+		if listErr := retryFilerOp("repointLatestBeforeDeletion.list", func() error {
+			var lerr error
+			entries, isLast, lerr = s3a.list(versionsDir, "", startFrom, false, filer.PaginationSize)
+			return lerr
+		}); listErr != nil {
+			return false, false, fmt.Errorf("list %s: %w", versionsDir, listErr)
+		}
+		for _, entry := range entries {
+			if entry == nil || entry.Extended == nil {
+				continue
+			}
+			vidBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]
+			if !ok {
+				continue
+			}
+			vid := string(vidBytes)
+			if vid == versionIdToDelete {
+				continue
+			}
+			if newLatestVersionId == "" || compareVersionIds(vid, newLatestVersionId) < 0 {
+				newLatestEntry = entry
+				newLatestVersionId = vid
+				newLatestVersionFile = entry.Name
+				newLatestIsDeleteMark = string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true"
+			}
+		}
+		if isLast || len(entries) == 0 {
+			break
+		}
+		startFrom = entries[len(entries)-1].Name
+	}
+
+	// Re-fetch the .versions/ entry so the CAS persist below sees the
+	// most recent Extended state. Distinguish NotFound (directory
+	// already gone — vacuous consistency, skip the post-step) from
+	// transient errors (must surface so the caller aborts before the
+	// blob rm — otherwise we'd remove the blob without having updated
+	// the pointer, reproducing the very dangling-state this PR fixes).
+	var versionsEntry *filer_pb.Entry
+	if getErr := retryFilerOp("repointLatestBeforeDeletion.getEntry", func() error {
+		var gerr error
+		versionsEntry, gerr = s3a.getEntry(bucketDir, versionsObjectPath)
+		return gerr
+	}); getErr != nil {
+		if errors.Is(getErr, filer_pb.ErrNotFound) || status.Code(getErr) == codes.NotFound {
+			// Directory already gone. Pointer is vacuously consistent.
+			return true, false, nil
+		}
+		return false, false, fmt.Errorf("read .versions entry: %w", getErr)
+	}
+	if versionsEntry.Extended == nil {
+		return false, false, nil
+	}
+	currentLatestIdBytes, hasCurrent := versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey]
+	if !hasCurrent || string(currentLatestIdBytes) != versionIdToDelete {
+		// A concurrent writer already moved the pointer; our delete is
+		// no longer the latest. Caller falls back to the existing path.
+		return false, false, nil
+	}
+
+	if newLatestEntry != nil {
+		// Multi-version: repoint to the prior latest.
+		versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(newLatestVersionId)
+		versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(newLatestVersionFile)
+		setCachedListMetadata(versionsEntry, newLatestEntry)
+		glog.V(2).Infof("repointLatestBeforeDeletion: %s/%s pre-roll to %s (deleteMarker=%v) before deleting %s", bucket, normalizedObject, newLatestVersionId, newLatestIsDeleteMark, versionIdToDelete)
+	} else {
+		// Singleton: clear the pointer fields. The blob rm and the
+		// post-rm .versions/ teardown follow in the caller; if those
+		// fail halfway through, the pointer is already absent so
+		// reads fall through to a clean NoSuchKey without entering the
+		// 10-retry stale-pointer self-heal path.
+		delete(versionsEntry.Extended, s3_constants.ExtLatestVersionIdKey)
+		delete(versionsEntry.Extended, s3_constants.ExtLatestVersionFileNameKey)
+		clearCachedVersionMetadata(versionsEntry.Extended)
+		glog.V(2).Infof("repointLatestBeforeDeletion: %s/%s clearing singleton pointer before deleting %s", bucket, normalizedObject, versionIdToDelete)
+	}
+
+	if mkErr := retryFilerOp("repointLatestBeforeDeletion.mkFile", func() error {
+		return s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
+			updatedEntry.Extended = versionsEntry.Extended
+			updatedEntry.Attributes = versionsEntry.Attributes
+			updatedEntry.Chunks = versionsEntry.Chunks
+		})
+	}); mkErr != nil {
+		return false, false, fmt.Errorf("persist repointed pointer: %w", mkErr)
+	}
+
+	return true, newLatestEntry == nil, nil
+}
+
+// retryAttempts and retryStep tune the bounded retries used when the
+// load-bearing filer ops in updateLatestVersionAfterDeletion fail with
+// transient errors. Doubled per attempt, capped at retryCap. Total
+// worst-case wall time ≈ 6.3s before propagating.
+const (
+	updateLatestRetryAttempts = 6
+	updateLatestRetryStep     = 100 * time.Millisecond
+	updateLatestRetryCap      = 2 * time.Second
+)
+
+// isRetryableFilerErr reports whether err is worth retrying through
+// retryFilerOp. Terminal conditions return false so the caller surfaces
+// them immediately without the backoff delay or the retry-budget
+// wrapper:
+//
+//   - NotFound: the entry genuinely doesn't exist. Retrying won't make
+//     it appear, and callers (e.g. repointLatestBeforeDeletion) want
+//     to act on this directly.
+//   - context.Canceled / DeadlineExceeded: the request was aborted by
+//     the client or hit a deadline. Continuing to retry just delays
+//     the failure return.
+//
+// Everything else (gRPC Unavailable, transient network errors, filer
+// overload signals, etc.) is treated as retryable.
+func isRetryableFilerErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound {
+		return false
+	}
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+	return true
+}
+
+func retryFilerOp(name string, fn func() error) error {
+	var lastErr error
+	backoff := updateLatestRetryStep
+	for attempt := 1; attempt <= updateLatestRetryAttempts; attempt++ {
+		err := fn()
+		if err == nil {
+			if attempt > 1 {
+				glog.V(1).Infof("retryFilerOp: %s succeeded on attempt %d", name, attempt)
+			}
+			return nil
+		}
+		if !isRetryableFilerErr(err) {
+			// Terminal — return raw so callers can errors.Is /
+			// status.Code on the unwrapped error and avoid the
+			// retry-budget delay.
+			return err
+		}
+		lastErr = err
+		if attempt == updateLatestRetryAttempts {
+			break
+		}
+		time.Sleep(backoff)
+		backoff *= 2
+		if backoff > updateLatestRetryCap {
+			backoff = updateLatestRetryCap
+		}
+	}
+	return fmt.Errorf("%s exhausted %d retries: %w", name, updateLatestRetryAttempts, lastErr)
+}
+
 // updateLatestVersionAfterDeletion finds the new latest version after deleting
 // the current latest. The pointer may refer to a delete marker: if a delete
 // marker is chronologically newer than the most recent remaining content
 // version, S3 semantics treat the object as deleted and the pointer must
 // reflect that. Restricting the scan to content versions here would resurrect
 // the object by promoting an older content version over a newer delete marker.
+//
+// All load-bearing filer interactions (list, getEntry, mkFile, rm) are
+// retried with bounded backoff. The function now returns a non-nil error
+// when those retries are exhausted; the caller is expected to surface the
+// failure (log + enqueue for the reconciler) instead of swallowing it,
+// which was the historic behaviour that left dangling pointers in place.
 func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(bucket, object string) error {
 	bucketDir := s3a.bucketDir(bucket)
 	versionsObjectPath := object + s3_constants.VersionsFolder
@@ -1097,10 +1350,17 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(bucket, object string)
 		startFrom             string
 	)
 	for {
-		entries, isLast, err := s3a.list(versionsDir, "", startFrom, false, filer.PaginationSize)
-		if err != nil {
+		var (
+			entries []*filer_pb.Entry
+			isLast  bool
+		)
+		if err := retryFilerOp("updateLatestVersionAfterDeletion.list", func() error {
+			var listErr error
+			entries, isLast, listErr = s3a.list(versionsDir, "", startFrom, false, filer.PaginationSize)
+			return listErr
+		}); err != nil {
 			glog.Errorf("updateLatestVersionAfterDeletion: failed to list versions in %s: %v", versionsDir, err)
-			return fmt.Errorf("failed to list versions: %v", err)
+			return fmt.Errorf("failed to list versions: %w", err)
 		}
 		totalEntries += len(entries)
 		if pageEntry, pageId, pageFile, pageDM := selectLatestVersion(entries); pageEntry != nil {
@@ -1120,9 +1380,13 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(bucket, object string)
 	glog.V(1).Infof("updateLatestVersionAfterDeletion: scanned %d entries in %s", totalEntries, versionsDir)
 
 	// Update the .versions directory metadata
-	versionsEntry, err := s3a.getEntry(bucketDir, versionsObjectPath)
-	if err != nil {
-		return fmt.Errorf("failed to get .versions directory: %v", err)
+	var versionsEntry *filer_pb.Entry
+	if err := retryFilerOp("updateLatestVersionAfterDeletion.getEntry", func() error {
+		var getErr error
+		versionsEntry, getErr = s3a.getEntry(bucketDir, versionsObjectPath)
+		return getErr
+	}); err != nil {
+		return fmt.Errorf("failed to get .versions directory: %w", err)
 	}
 
 	if versionsEntry.Extended == nil {
@@ -1137,13 +1401,14 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(bucket, object string)
 		setCachedListMetadata(versionsEntry, latestVersionEntry)
 
 		glog.V(2).Infof("updateLatestVersionAfterDeletion: new latest version for %s/%s is %s (deleteMarker=%v)", bucket, object, latestVersionId, latestIsDeleteMarker)
-		err = s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
-			updatedEntry.Extended = versionsEntry.Extended
-			updatedEntry.Attributes = versionsEntry.Attributes
-			updatedEntry.Chunks = versionsEntry.Chunks
-		})
-		if err != nil {
-			return fmt.Errorf("failed to update .versions directory metadata: %v", err)
+		if err := retryFilerOp("updateLatestVersionAfterDeletion.mkFile", func() error {
+			return s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
+				updatedEntry.Extended = versionsEntry.Extended
+				updatedEntry.Attributes = versionsEntry.Attributes
+				updatedEntry.Chunks = versionsEntry.Chunks
+			})
+		}); err != nil {
+			return fmt.Errorf("failed to update .versions directory metadata: %w", err)
 		}
 	} else {
 		// No version-tagged entries remain - try to delete the .versions
@@ -1161,12 +1426,36 @@ func (s3a *S3ApiServer) updateLatestVersionAfterDeletion(bucket, object string)
 		// object is correctly absent.
 		glog.V(2).Infof("updateLatestVersionAfterDeletion: no versions left for %s/%s, deleting .versions directory", bucket, object)
 
-		err = s3a.rm(bucketDir, versionsObjectPath, true, false)
-		if err == nil {
+		rmErr := s3a.rm(bucketDir, versionsObjectPath, true, false)
+		if rmErr == nil {
 			return nil
 		}
-		glog.Warningf("updateLatestVersionAfterDeletion: failed to delete .versions directory for %s/%s: %v", bucket, object, err)
+		// Two ways rm can fail here: "non-empty folder" (orphan entries
+		// blocking the teardown — fall through to pointer clear) and a
+		// transient filer error (worth retrying). Distinguish by the
+		// canonical error substring; if we can't tell, treat as transient.
+		if strings.Contains(rmErr.Error(), filer.MsgFailDelNonEmptyFolder) {
+			glog.V(2).Infof("updateLatestVersionAfterDeletion: .versions/ for %s/%s still has orphan entries: %v", bucket, object, rmErr)
+			s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
+			return nil
+		}
+		// Transient — retry the rm a few times before giving up. Even
+		// if it ultimately fails, we still clear the stale pointer so
+		// readers get a clean miss; the directory can be tidied by the
+		// reconciler later.
+		retryErr := retryFilerOp("updateLatestVersionAfterDeletion.rm", func() error {
+			return s3a.rm(bucketDir, versionsObjectPath, true, false)
+		})
+		if retryErr == nil {
+			return nil
+		}
+		if strings.Contains(retryErr.Error(), filer.MsgFailDelNonEmptyFolder) {
+			s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
+			return nil
+		}
+		glog.Warningf("updateLatestVersionAfterDeletion: failed to delete .versions directory for %s/%s after retries: %v", bucket, object, retryErr)
 		s3a.clearStaleLatestVersionPointer(bucket, object, bucketDir, versionsObjectPath, versionsEntry, "updateLatestVersionAfterDeletion")
+		return fmt.Errorf("delete .versions directory: %w", retryErr)
 	}
 
 	return nil

weed/s3api/s3api_server.go

@@ -101,6 +101,9 @@ type S3ApiServer struct {
 	// teardown waits on context.Background() fetches. The chunkCache field
 	// is nil in this commit; a follow-up wires in an in-memory chunk cache.
 	readerCache *filer.ReaderCache
+
+	versionsHealQueue       *versionsHealQueue
+	versionsReconcilerStop  func()
 }
 
 type objectWriteLock interface {
@@ -381,10 +384,17 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	// Start bucket size metrics collection in background
 	go s3ApiServer.startBucketSizeMetricsLoop(context.Background())
 
+	// Start the versioning reconciler that drains stranded .versions/
+	// pointer-to-missing-file states without waiting for a client GET.
+	s3ApiServer.versionsReconcilerStop = s3ApiServer.startVersioningReconciler()
+
 	return s3ApiServer, nil
 }
 
 func (s3a *S3ApiServer) Shutdown() {
+	if s3a.versionsReconcilerStop != nil {
+		s3a.versionsReconcilerStop()
+	}
 	if s3a.iam != nil {
 		s3a.iam.Shutdown()
 	}

weed/s3api/s3api_versioning_reconciler.go

@@ -0,0 +1,208 @@
+package s3api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// Background reconciler for .versions/ directories whose latest-version
+// pointer references a missing file. The inline delete path
+// (deleteSpecificObjectVersion -> updateLatestVersionAfterDeletion) can
+// leave this state behind if the second step (pointer update / dir rm)
+// fails after the first (blob rm) succeeded. Read-path self-heal already
+// recovers on the next GET, but that surfaces NoSuchKey to the client.
+// The reconciler drains stranded entries proactively so a passing health
+// check or an idle key gets repaired without waiting for a client read.
+
+const (
+	versionsHealQueueCapacity = 4096
+	versionsHealPollInterval  = 5 * time.Second
+	versionsHealMaxRetries    = 6
+	versionsHealBaseBackoff   = 200 * time.Millisecond
+	versionsHealMaxBackoff    = 30 * time.Second
+)
+
+type versionsHealCandidate struct {
+	bucket    string
+	object    string
+	enqueued  time.Time
+	attempts  int
+	nextRetry time.Time
+}
+
+type versionsHealQueue struct {
+	mu      sync.Mutex
+	pending map[string]*versionsHealCandidate
+}
+
+func newVersionsHealQueue() *versionsHealQueue {
+	return &versionsHealQueue{
+		pending: make(map[string]*versionsHealCandidate),
+	}
+}
+
+func versionsHealKey(bucket, object string) string {
+	return bucket + "/" + object
+}
+
+// Enqueue records that bucket/object's .versions/ directory may be in a
+// stranded state and needs the reconciler to verify and heal it. Bounded
+// by versionsHealQueueCapacity so a stuck filer can't grow the map
+// without limit; on overflow we drop the newest candidate (the heal will
+// run on the next read or the next delete-failure for that key).
+func (q *versionsHealQueue) Enqueue(bucket, object string) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	key := versionsHealKey(bucket, object)
+	if _, ok := q.pending[key]; ok {
+		return
+	}
+	if len(q.pending) >= versionsHealQueueCapacity {
+		glog.V(1).Infof("versionsHealQueue: capacity reached, dropping %s/%s", bucket, object)
+		return
+	}
+	q.pending[key] = &versionsHealCandidate{
+		bucket:    bucket,
+		object:    object,
+		enqueued:  time.Now(),
+		nextRetry: time.Now(),
+	}
+}
+
+// popReady returns candidates whose nextRetry is in the past, removing
+// them from the queue. Callers reinsert via requeue() if the heal failed.
+func (q *versionsHealQueue) popReady(now time.Time) []*versionsHealCandidate {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	var out []*versionsHealCandidate
+	for k, c := range q.pending {
+		if !c.nextRetry.After(now) {
+			out = append(out, c)
+			delete(q.pending, k)
+		}
+	}
+	return out
+}
+
+// requeue re-inserts a candidate with an updated retry schedule. Drops
+// candidates that have hit the attempt cap so they don't loop forever
+// against a deterministically broken state; the read-path heal remains
+// as a last-resort safety net.
+func (q *versionsHealQueue) requeue(c *versionsHealCandidate, backoff time.Duration) {
+	if c.attempts >= versionsHealMaxRetries {
+		glog.Warningf("versionsHealQueue: giving up on %s/%s after %d attempts (read-path heal will still recover)", c.bucket, c.object, c.attempts)
+		return
+	}
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	c.nextRetry = time.Now().Add(backoff)
+	q.pending[versionsHealKey(c.bucket, c.object)] = c
+}
+
+// Len returns the current queued size. Used in tests and by metrics.
+func (q *versionsHealQueue) Len() int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return len(q.pending)
+}
+
+// startVersioningReconciler launches the background worker. Returns a
+// stop function that the server's Shutdown calls to cancel the loop.
+func (s3a *S3ApiServer) startVersioningReconciler() (stop func()) {
+	if s3a.versionsHealQueue == nil {
+		s3a.versionsHealQueue = newVersionsHealQueue()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go s3a.runVersioningReconciler(ctx)
+	return cancel
+}
+
+func (s3a *S3ApiServer) runVersioningReconciler(ctx context.Context) {
+	ticker := time.NewTicker(versionsHealPollInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s3a.drainVersionsHealQueue()
+		}
+	}
+}
+
+func (s3a *S3ApiServer) drainVersionsHealQueue() {
+	q := s3a.versionsHealQueue
+	if q == nil {
+		return
+	}
+	ready := q.popReady(time.Now())
+	for _, c := range ready {
+		c.attempts++
+		if err := s3a.healVersionsPointer(c.bucket, c.object); err != nil {
+			backoff := versionsHealBaseBackoff << uint(c.attempts-1)
+			if backoff > versionsHealMaxBackoff {
+				backoff = versionsHealMaxBackoff
+			}
+			glog.V(1).Infof("versionsHealQueue: heal failed for %s/%s (attempt %d): %v; retry in %v", c.bucket, c.object, c.attempts, err, backoff)
+			q.requeue(c, backoff)
+			continue
+		}
+		glog.V(1).Infof("versionsHealQueue: healed %s/%s after %d attempt(s) (queued for %v)", c.bucket, c.object, c.attempts, time.Since(c.enqueued))
+	}
+}
+
+// healVersionsPointer reads the .versions directory entry for the given
+// object and, if its latest-version pointer references a file that does
+// not exist, invokes the same self-heal path used by the read flow. A
+// genuinely-missing .versions directory or an already-consistent pointer
+// is treated as success — the candidate either healed itself via another
+// path or was never in trouble.
+//
+// Transient errors from the filer (timeout, brief unreachability) are
+// returned so the reconciler retries with backoff rather than silently
+// evicting the candidate. Swallowing them as "no stranded state" would
+// defeat the reconciler for exactly the failure modes the PR targets.
+func (s3a *S3ApiServer) healVersionsPointer(bucket, object string) error {
+	bucketDir := s3a.bucketDir(bucket)
+	versionsObjectPath := object + s3_constants.VersionsFolder
+	versionsEntry, err := s3a.getEntry(bucketDir, versionsObjectPath)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound {
+			// Directory is gone — no stranded state to heal.
+			return nil
+		}
+		return fmt.Errorf("read .versions entry %s/%s: %w", bucket, object, err)
+	}
+	if versionsEntry == nil || versionsEntry.Extended == nil {
+		return nil
+	}
+	fileBytes, hasFile := versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]
+	if !hasFile || len(fileBytes) == 0 {
+		// No pointer => no stranded state.
+		return nil
+	}
+	latestFile := string(fileBytes)
+	if _, probeErr := s3a.getEntry(bucketDir+"/"+versionsObjectPath, latestFile); probeErr == nil {
+		// Pointer is consistent.
+		return nil
+	} else if !errors.Is(probeErr, filer_pb.ErrNotFound) && status.Code(probeErr) != codes.NotFound {
+		// Transient — surface so the reconciler retries with backoff
+		// rather than invoking heal on a possibly-incomplete listing
+		// (which could rewrite the pointer to an older version).
+		return fmt.Errorf("probe latest version %s/%s/%s: %w", bucket, object, latestFile, probeErr)
+	}
+	// Reuse the read-path heal so behaviour stays identical: it will
+	// repair the pointer to the newest remaining version, or clear it
+	// when nothing valid remains.
+	_, healErr := s3a.healStaleLatestVersionPointer(bucket, object, versionsEntry, latestFile)
+	return healErr
+}

weed/s3api/s3api_versioning_reconciler_test.go

@@ -0,0 +1,144 @@
+package s3api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// TestVersionsHealQueue_DedupOnEnqueue ensures multiple enqueues of the
+// same bucket/object collapse into a single pending entry, so a hot
+// failure path doesn't bloat the queue.
+func TestVersionsHealQueue_DedupOnEnqueue(t *testing.T) {
+	q := newVersionsHealQueue()
+	for i := 0; i < 5; i++ {
+		q.Enqueue("b", "obj")
+	}
+	assert.Equal(t, 1, q.Len(), "duplicate enqueues collapse")
+}
+
+// TestVersionsHealQueue_CapacityCap ensures the queue refuses growth
+// past the static cap and logs at V(1) instead of OOM-ing.
+func TestVersionsHealQueue_CapacityCap(t *testing.T) {
+	q := newVersionsHealQueue()
+	for i := 0; i < versionsHealQueueCapacity+50; i++ {
+		q.Enqueue("b", string(rune(i)))
+	}
+	assert.Equal(t, versionsHealQueueCapacity, q.Len(), "queue clamps at capacity")
+}
+
+// TestVersionsHealQueue_PopReadyOnlyDueItems checks that nextRetry
+// gating keeps not-yet-ready candidates in the queue.
+func TestVersionsHealQueue_PopReadyOnlyDueItems(t *testing.T) {
+	q := newVersionsHealQueue()
+	q.Enqueue("b", "due")
+
+	// Inject a deferred candidate directly so we control its nextRetry.
+	q.pending[versionsHealKey("b", "later")] = &versionsHealCandidate{
+		bucket:    "b",
+		object:    "later",
+		enqueued:  time.Now(),
+		nextRetry: time.Now().Add(10 * time.Minute),
+	}
+
+	due := q.popReady(time.Now())
+	require.Len(t, due, 1, "only the due candidate pops")
+	assert.Equal(t, "due", due[0].object)
+	assert.Equal(t, 1, q.Len(), "deferred candidate still queued")
+}
+
+// TestVersionsHealQueue_RequeueWithBackoff verifies failed candidates
+// re-enter the queue with an extended nextRetry.
+func TestVersionsHealQueue_RequeueWithBackoff(t *testing.T) {
+	q := newVersionsHealQueue()
+	c := &versionsHealCandidate{bucket: "b", object: "obj", attempts: 1}
+
+	q.requeue(c, 500*time.Millisecond)
+	assert.Equal(t, 1, q.Len())
+
+	now := time.Now()
+	due := q.popReady(now)
+	assert.Empty(t, due, "not yet due immediately after requeue")
+
+	due = q.popReady(now.Add(time.Second))
+	assert.Len(t, due, 1, "due after backoff window passes")
+}
+
+// TestVersionsHealQueue_GiveUpAfterMaxAttempts ensures we don't loop
+// forever against a deterministically broken state.
+func TestVersionsHealQueue_GiveUpAfterMaxAttempts(t *testing.T) {
+	q := newVersionsHealQueue()
+	c := &versionsHealCandidate{bucket: "b", object: "obj", attempts: versionsHealMaxRetries}
+	q.requeue(c, time.Millisecond)
+	assert.Equal(t, 0, q.Len(), "candidate at max attempts is dropped, read-path heal still covers it")
+}
+
+// TestRetryFilerOp_SucceedsBeforeExhaustion confirms a flaky op that
+// eventually succeeds is reported as success without surfacing prior
+// errors.
+func TestRetryFilerOp_SucceedsBeforeExhaustion(t *testing.T) {
+	calls := 0
+	err := retryFilerOp("test", func() error {
+		calls++
+		if calls < 3 {
+			return errors.New("transient")
+		}
+		return nil
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, 3, calls, "stops calling once the op succeeds")
+}
+
+// TestRetryFilerOp_PropagatesAfterExhaustion confirms a deterministic
+// failure is wrapped with the attempt count so operators can tell at
+// a glance whether the underlying issue is transient.
+func TestRetryFilerOp_PropagatesAfterExhaustion(t *testing.T) {
+	calls := 0
+	err := retryFilerOp("test", func() error {
+		calls++
+		return errors.New("permanent")
+	})
+	require.Error(t, err)
+	assert.Equal(t, updateLatestRetryAttempts, calls, "ran the full retry budget")
+	assert.Contains(t, err.Error(), "exhausted")
+	assert.Contains(t, err.Error(), "permanent", "underlying error preserved")
+}
+
+// TestRetryFilerOp_TerminalErrorsShortCircuit confirms that errors which
+// won't change on retry (NotFound, context cancellation) are returned
+// immediately and unwrapped, without burning the retry budget.
+func TestRetryFilerOp_TerminalErrorsShortCircuit(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{"filer_pb.ErrNotFound", filer_pb.ErrNotFound},
+		{"wrapped filer_pb.ErrNotFound", fmt.Errorf("wrap: %w", filer_pb.ErrNotFound)},
+		{"grpc NotFound status", status.Error(codes.NotFound, "missing")},
+		{"context canceled", context.Canceled},
+		{"context deadline exceeded", context.DeadlineExceeded},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			start := time.Now()
+			calls := 0
+			err := retryFilerOp("test", func() error {
+				calls++
+				return tc.err
+			})
+			elapsed := time.Since(start)
+			require.Error(t, err)
+			assert.Equal(t, 1, calls, "terminal error must not be retried")
+			assert.Less(t, elapsed, 50*time.Millisecond, "terminal error must not delay")
+			assert.NotContains(t, err.Error(), "exhausted", "terminal error must not be wrapped with retry-budget prefix")
+		})
+	}
+}