diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 487a27135..a74b9466e 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/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 diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 7fb37f250..18f141c27 100644 --- a/weed/s3api/s3api_server.go +++ b/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() } diff --git a/weed/s3api/s3api_versioning_reconciler.go b/weed/s3api/s3api_versioning_reconciler.go new file mode 100644 index 000000000..1a64b2d2f --- /dev/null +++ b/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 +} diff --git a/weed/s3api/s3api_versioning_reconciler_test.go b/weed/s3api/s3api_versioning_reconciler_test.go new file mode 100644 index 000000000..1f890f7a5 --- /dev/null +++ b/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") + }) + } +}