diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 71185d3d1..b86ac3c5b 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -54,6 +54,8 @@ type Filer struct { RemoteStorage *FilerRemoteStorage Dlm *lock_manager.DistributedLockManager MaxFilenameLength uint32 + deletionQuit chan struct{} + DeletionRetryQueue *DeletionRetryQueue } func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer { @@ -66,6 +68,8 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH UniqueFilerId: util.RandomInt32(), Dlm: lock_manager.NewDistributedLockManager(filerHost), MaxFilenameLength: maxFilenameLength, + deletionQuit: make(chan struct{}), + DeletionRetryQueue: NewDeletionRetryQueue(), } if f.UniqueFilerId < 0 { f.UniqueFilerId = -f.UniqueFilerId @@ -379,6 +383,7 @@ func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, sta } func (f *Filer) Shutdown() { + close(f.deletionQuit) f.LocalMetaLogBuffer.ShutdownLogBuffer() f.Store.Shutdown() } diff --git a/weed/filer/filer_deletion.go b/weed/filer/filer_deletion.go index b3a4296ba..d8bc105e6 100644 --- a/weed/filer/filer_deletion.go +++ b/weed/filer/filer_deletion.go @@ -1,20 +1,274 @@ package filer import ( + "container/heap" "context" "fmt" "strings" + "sync" "time" + "google.golang.org/grpc" + "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/wdclient" ) +const ( + // Maximum number of retry attempts for failed deletions + MaxRetryAttempts = 10 + // Initial retry delay (will be doubled with each attempt) + InitialRetryDelay = 5 * time.Minute + // Maximum retry delay + MaxRetryDelay = 6 * time.Hour + // Interval for checking retry queue for ready items + DeletionRetryPollInterval = 1 * time.Minute + // Maximum number of items to process per retry iteration + DeletionRetryBatchSize = 1000 + // Maximum number of error details to include in log messages + MaxLoggedErrorDetails = 10 + // Interval for polling the deletion queue for new items + // Using a prime number to de-synchronize with other periodic tasks + DeletionPollInterval = 1123 * time.Millisecond + // Maximum number of file IDs to delete per batch (roughly 20 bytes per file ID) + DeletionBatchSize = 100000 +) + +// retryablePatterns contains error message patterns that indicate temporary/transient conditions +// that should be retried. These patterns are based on actual error messages from the deletion pipeline. +var retryablePatterns = []string{ + "is read only", // Volume temporarily read-only (tiering, maintenance) + "error reading from server", // Network I/O errors + "connection reset by peer", // Network connection issues + "closed network connection", // Network connection closed unexpectedly + "connection refused", // Server temporarily unavailable + "timeout", // Operation timeout (network or server) + "deadline exceeded", // Context deadline exceeded + "context canceled", // Context cancellation (may be transient) + "lookup error", // Volume lookup failures + "lookup failed", // Volume server discovery issues + "too many requests", // Rate limiting / backpressure + "service unavailable", // HTTP 503 errors + "temporarily unavailable", // Temporary service issues + "try again", // Explicit retry suggestion + "i/o timeout", // Network I/O timeout + "broken pipe", // Connection broken during operation +} + +// DeletionRetryItem represents a file deletion that failed and needs to be retried +type DeletionRetryItem struct { + FileId string + RetryCount int + NextRetryAt time.Time + LastError string + heapIndex int // index in the heap (for heap.Interface) + inFlight bool // true when item is being processed, prevents duplicate additions +} + +// retryHeap implements heap.Interface for DeletionRetryItem +// Items are ordered by NextRetryAt (earliest first) +type retryHeap []*DeletionRetryItem + +// Compile-time assertion that retryHeap implements heap.Interface +var _ heap.Interface = (*retryHeap)(nil) + +func (h retryHeap) Len() int { return len(h) } + +func (h retryHeap) Less(i, j int) bool { + return h[i].NextRetryAt.Before(h[j].NextRetryAt) +} + +func (h retryHeap) Swap(i, j int) { + h[i], h[j] = h[j], h[i] + h[i].heapIndex = i + h[j].heapIndex = j +} + +func (h *retryHeap) Push(x any) { + item := x.(*DeletionRetryItem) + item.heapIndex = len(*h) + *h = append(*h, item) +} + +func (h *retryHeap) Pop() any { + old := *h + n := len(old) + item := old[n-1] + old[n-1] = nil // avoid memory leak + item.heapIndex = -1 // mark as removed + *h = old[0 : n-1] + return item +} + +// DeletionRetryQueue manages the queue of failed deletions that need to be retried. +// Uses a min-heap ordered by NextRetryAt for efficient retrieval of ready items. +// +// LIMITATION: Current implementation stores retry queue in memory only. +// On filer restart, all pending retries are lost. With MaxRetryDelay up to 6 hours, +// process restarts during this window will cause retry state loss. +// +// TODO: Consider persisting retry queue to durable storage for production resilience: +// - Option 1: Leverage existing Filer store (KV operations) +// - Option 2: Periodic snapshots to disk with recovery on startup +// - Option 3: Write-ahead log for retry queue mutations +// - Trade-offs: Performance vs durability, complexity vs reliability +// +// For now, accepting in-memory storage as pragmatic initial implementation. +// Lost retries will be eventually consistent as files remain in deletion queue. +type DeletionRetryQueue struct { + heap retryHeap + itemIndex map[string]*DeletionRetryItem // for O(1) lookup by FileId + lock sync.Mutex +} + +// NewDeletionRetryQueue creates a new retry queue +func NewDeletionRetryQueue() *DeletionRetryQueue { + q := &DeletionRetryQueue{ + heap: make(retryHeap, 0), + itemIndex: make(map[string]*DeletionRetryItem), + } + heap.Init(&q.heap) + return q +} + +// calculateBackoff calculates the exponential backoff delay for a given retry count. +// Uses exponential backoff formula: InitialRetryDelay * 2^(retryCount-1) +// The first retry (retryCount=1) uses InitialRetryDelay, second uses 2x, third uses 4x, etc. +// Includes overflow protection and caps at MaxRetryDelay. +func calculateBackoff(retryCount int) time.Duration { + // The first retry is attempt 1, but shift should start at 0 + if retryCount <= 1 { + return InitialRetryDelay + } + + shiftAmount := uint(retryCount - 1) + + // time.Duration is an int64. A left shift of 63 or more will result in a + // negative number or zero. The multiplication can also overflow much earlier + // (around a shift of 25 for a 5-minute initial delay). + // The `delay <= 0` check below correctly catches all these overflow cases. + delay := InitialRetryDelay << shiftAmount + + if delay <= 0 || delay > MaxRetryDelay { + return MaxRetryDelay + } + + return delay +} + +// AddOrUpdate adds a new failed deletion or updates an existing one +// Time complexity: O(log N) for insertion/update +func (q *DeletionRetryQueue) AddOrUpdate(fileId string, errorMsg string) { + q.lock.Lock() + defer q.lock.Unlock() + + // Check if item already exists (including in-flight items) + if item, exists := q.itemIndex[fileId]; exists { + // Item is already in the queue or being processed. Just update the error. + // The existing retry schedule should proceed. + // RetryCount is only incremented in RequeueForRetry when an actual retry is performed. + item.LastError = errorMsg + if item.inFlight { + glog.V(2).Infof("retry for %s in-flight: attempt %d, will preserve retry state", fileId, item.RetryCount) + } else { + glog.V(2).Infof("retry for %s already scheduled: attempt %d, next retry in %v", fileId, item.RetryCount, time.Until(item.NextRetryAt)) + } + return + } + + // Add new item + delay := InitialRetryDelay + item := &DeletionRetryItem{ + FileId: fileId, + RetryCount: 1, + NextRetryAt: time.Now().Add(delay), + LastError: errorMsg, + inFlight: false, + } + heap.Push(&q.heap, item) + q.itemIndex[fileId] = item + glog.V(2).Infof("added retry for %s: next retry in %v", fileId, delay) +} + +// RequeueForRetry re-adds a previously failed item back to the queue with incremented retry count. +// This method MUST be used when re-queuing items from processRetryBatch to preserve retry state. +// Time complexity: O(log N) for insertion +func (q *DeletionRetryQueue) RequeueForRetry(item *DeletionRetryItem, errorMsg string) { + q.lock.Lock() + defer q.lock.Unlock() + + // Increment retry count + item.RetryCount++ + item.LastError = errorMsg + + // Calculate next retry time with exponential backoff + delay := calculateBackoff(item.RetryCount) + item.NextRetryAt = time.Now().Add(delay) + item.inFlight = false // Clear in-flight flag + glog.V(2).Infof("requeued retry for %s: attempt %d, next retry in %v", item.FileId, item.RetryCount, delay) + + // Re-add to heap (item still in itemIndex) + heap.Push(&q.heap, item) +} + +// GetReadyItems returns items that are ready to be retried and marks them as in-flight +// Time complexity: O(K log N) where K is the number of ready items +// Items are processed in order of NextRetryAt (earliest first) +func (q *DeletionRetryQueue) GetReadyItems(maxItems int) []*DeletionRetryItem { + q.lock.Lock() + defer q.lock.Unlock() + + now := time.Now() + var readyItems []*DeletionRetryItem + + // Peek at items from the top of the heap (earliest NextRetryAt) + for len(q.heap) > 0 && len(readyItems) < maxItems { + item := q.heap[0] + + // If the earliest item is not ready yet, no other items are ready either + if item.NextRetryAt.After(now) { + break + } + + // Remove from heap but keep in itemIndex with inFlight flag + heap.Pop(&q.heap) + + if item.RetryCount <= MaxRetryAttempts { + item.inFlight = true // Mark as being processed + readyItems = append(readyItems, item) + } else { + // Max attempts reached, log and discard completely + delete(q.itemIndex, item.FileId) + glog.Warningf("max retry attempts (%d) reached for %s, last error: %s", MaxRetryAttempts, item.FileId, item.LastError) + } + } + + return readyItems +} + +// Remove removes an item from the queue (called when deletion succeeds or fails permanently) +// Time complexity: O(1) +func (q *DeletionRetryQueue) Remove(item *DeletionRetryItem) { + q.lock.Lock() + defer q.lock.Unlock() + + // Item was already removed from heap by GetReadyItems, just remove from index + delete(q.itemIndex, item.FileId) +} + +// Size returns the current size of the retry queue +func (q *DeletionRetryQueue) Size() int { + q.lock.Lock() + defer q.lock.Unlock() + return len(q.heap) +} + func LookupByMasterClientFn(masterClient *wdclient.MasterClient) func(vids []string) (map[string]*operation.LookupResult, error) { return func(vids []string) (map[string]*operation.LookupResult, error) { m := make(map[string]*operation.LookupResult) @@ -41,64 +295,292 @@ func (f *Filer) loopProcessingDeletion() { lookupFunc := LookupByMasterClientFn(f.MasterClient) - DeletionBatchSize := 100000 // roughly 20 bytes cost per file id. + // Start retry processor in a separate goroutine + go f.loopProcessingDeletionRetry(lookupFunc) + + ticker := time.NewTicker(DeletionPollInterval) + defer ticker.Stop() - var deletionCount int for { - deletionCount = 0 - f.fileIdDeletionQueue.Consume(func(fileIds []string) { - for len(fileIds) > 0 { - var toDeleteFileIds []string - if len(fileIds) > DeletionBatchSize { - toDeleteFileIds = fileIds[:DeletionBatchSize] - fileIds = fileIds[DeletionBatchSize:] - } else { - toDeleteFileIds = fileIds - fileIds = fileIds[:0] - } - deletionCount = len(toDeleteFileIds) - results := operation.DeleteFileIdsWithLookupVolumeId(f.GrpcDialOption, toDeleteFileIds, lookupFunc) - - // Process individual results for better error tracking - var successCount, notFoundCount, errorCount int - var errorDetails []string - - for _, result := range results { - if result.Error == "" { - successCount++ - } else if result.Error == "not found" || strings.Contains(result.Error, storage.ErrorDeleted.Error()) { - // Already deleted - acceptable - notFoundCount++ - } else { - // Actual error - errorCount++ - if errorCount <= 10 { - // Only log first 10 errors to avoid flooding logs - errorDetails = append(errorDetails, result.FileId+": "+result.Error) - } + select { + case <-f.deletionQuit: + glog.V(0).Infof("deletion processor shutting down") + return + case <-ticker.C: + f.fileIdDeletionQueue.Consume(func(fileIds []string) { + for i := 0; i < len(fileIds); i += DeletionBatchSize { + end := i + DeletionBatchSize + if end > len(fileIds) { + end = len(fileIds) } + toDeleteFileIds := fileIds[i:end] + f.processDeletionBatch(toDeleteFileIds, lookupFunc) } + }) + } + } +} - if successCount > 0 || notFoundCount > 0 { - glog.V(2).Infof("deleted %d files successfully, %d already deleted (not found)", successCount, notFoundCount) - } +// processDeletionBatch handles deletion of a batch of file IDs and processes results. +// It classifies errors into retryable and permanent categories, adds retryable failures +// to the retry queue, and logs appropriate messages. +func (f *Filer) processDeletionBatch(toDeleteFileIds []string, lookupFunc func([]string) (map[string]*operation.LookupResult, error)) { + // Deduplicate file IDs to prevent incorrect retry count increments for the same file ID within a single batch. + uniqueFileIdsSlice := make([]string, 0, len(toDeleteFileIds)) + processed := make(map[string]struct{}, len(toDeleteFileIds)) + for _, fileId := range toDeleteFileIds { + if _, found := processed[fileId]; !found { + processed[fileId] = struct{}{} + uniqueFileIdsSlice = append(uniqueFileIdsSlice, fileId) + } + } - if errorCount > 0 { - logMessage := fmt.Sprintf("failed to delete %d/%d files", errorCount, len(toDeleteFileIds)) - if errorCount > 10 { - logMessage += " (showing first 10)" - } - glog.V(0).Infof("%s: %v", logMessage, strings.Join(errorDetails, "; ")) + if len(uniqueFileIdsSlice) == 0 { + return + } + + // Delete files and classify outcomes + outcomes := deleteFilesAndClassify(f.GrpcDialOption, uniqueFileIdsSlice, lookupFunc) + + // Process outcomes + var successCount, notFoundCount, retryableErrorCount, permanentErrorCount int + var errorDetails []string + + for _, fileId := range uniqueFileIdsSlice { + outcome := outcomes[fileId] + + switch outcome.status { + case deletionOutcomeSuccess: + successCount++ + case deletionOutcomeNotFound: + notFoundCount++ + case deletionOutcomeRetryable, deletionOutcomeNoResult: + retryableErrorCount++ + f.DeletionRetryQueue.AddOrUpdate(fileId, outcome.errorMsg) + if len(errorDetails) < MaxLoggedErrorDetails { + errorDetails = append(errorDetails, fileId+": "+outcome.errorMsg+" (will retry)") + } + case deletionOutcomePermanent: + permanentErrorCount++ + if len(errorDetails) < MaxLoggedErrorDetails { + errorDetails = append(errorDetails, fileId+": "+outcome.errorMsg+" (permanent)") + } + } + } + + if successCount > 0 || notFoundCount > 0 { + glog.V(2).Infof("deleted %d files successfully, %d already deleted (not found)", successCount, notFoundCount) + } + + totalErrors := retryableErrorCount + permanentErrorCount + if totalErrors > 0 { + logMessage := fmt.Sprintf("failed to delete %d/%d files (%d retryable, %d permanent)", + totalErrors, len(uniqueFileIdsSlice), retryableErrorCount, permanentErrorCount) + if len(errorDetails) > 0 { + if totalErrors > MaxLoggedErrorDetails { + logMessage += fmt.Sprintf(" (showing first %d)", len(errorDetails)) + } + glog.V(0).Infof("%s: %v", logMessage, strings.Join(errorDetails, "; ")) + } else { + glog.V(0).Info(logMessage) + } + } + + if f.DeletionRetryQueue.Size() > 0 { + glog.V(2).Infof("retry queue size: %d", f.DeletionRetryQueue.Size()) + } +} + +const ( + deletionOutcomeSuccess = "success" + deletionOutcomeNotFound = "not_found" + deletionOutcomeRetryable = "retryable" + deletionOutcomePermanent = "permanent" + deletionOutcomeNoResult = "no_result" +) + +// deletionOutcome represents the result of classifying deletion results for a file +type deletionOutcome struct { + status string // One of the deletionOutcome* constants + errorMsg string +} + +// deleteFilesAndClassify performs deletion and classifies outcomes for a list of file IDs +func deleteFilesAndClassify(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func([]string) (map[string]*operation.LookupResult, error)) map[string]deletionOutcome { + // Perform deletion + results := operation.DeleteFileIdsWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc) + + // Group results by file ID to handle multiple results for replicated volumes + resultsByFileId := make(map[string][]*volume_server_pb.DeleteResult) + for _, result := range results { + resultsByFileId[result.FileId] = append(resultsByFileId[result.FileId], result) + } + + // Classify outcome for each file + outcomes := make(map[string]deletionOutcome, len(fileIds)) + for _, fileId := range fileIds { + outcomes[fileId] = classifyDeletionOutcome(fileId, resultsByFileId) + } + + return outcomes +} + +// classifyDeletionOutcome examines all deletion results for a file ID and determines the overall outcome +// Uses a single pass through results with early return for permanent errors (highest priority) +// Priority: Permanent > Retryable > Success > Not Found +func classifyDeletionOutcome(fileId string, resultsByFileId map[string][]*volume_server_pb.DeleteResult) deletionOutcome { + fileIdResults, found := resultsByFileId[fileId] + if !found || len(fileIdResults) == 0 { + return deletionOutcome{ + status: deletionOutcomeNoResult, + errorMsg: "no deletion result from volume server", + } + } + + var firstRetryableError string + hasSuccess := false + + for _, res := range fileIdResults { + if res.Error == "" { + hasSuccess = true + continue + } + if strings.Contains(res.Error, storage.ErrorDeleted.Error()) || res.Error == "not found" { + continue + } + + if isRetryableError(res.Error) { + if firstRetryableError == "" { + firstRetryableError = res.Error + } + } else { + // Permanent error takes highest precedence - return immediately + return deletionOutcome{status: deletionOutcomePermanent, errorMsg: res.Error} + } + } + + if firstRetryableError != "" { + return deletionOutcome{status: deletionOutcomeRetryable, errorMsg: firstRetryableError} + } + + if hasSuccess { + return deletionOutcome{status: deletionOutcomeSuccess, errorMsg: ""} + } + + // If we are here, all results were "not found" + return deletionOutcome{status: deletionOutcomeNotFound, errorMsg: ""} +} + +// isRetryableError determines if an error is retryable based on its message. +// +// Current implementation uses string matching which is brittle and may break +// if error messages change in dependencies. This is acceptable for the initial +// implementation but should be improved in the future. +// +// TODO: Consider these improvements for more robust error handling: +// - Pass DeleteResult instead of just error string to access Status codes +// - Use HTTP status codes (503 Service Unavailable, 429 Too Many Requests, etc.) +// - Implement structured error types that can be checked with errors.Is/errors.As +// - Extract and check gRPC status codes for better classification +// - Add error wrapping in the deletion pipeline to preserve error context +// +// For now, we use conservative string matching for known transient error patterns. +func isRetryableError(errorMsg string) bool { + // Empty errors are not retryable + if errorMsg == "" { + return false + } + + errorLower := strings.ToLower(errorMsg) + for _, pattern := range retryablePatterns { + if strings.Contains(errorLower, pattern) { + return true + } + } + return false +} + +// loopProcessingDeletionRetry processes the retry queue for failed deletions +func (f *Filer) loopProcessingDeletionRetry(lookupFunc func([]string) (map[string]*operation.LookupResult, error)) { + + ticker := time.NewTicker(DeletionRetryPollInterval) + defer ticker.Stop() + + for { + select { + case <-f.deletionQuit: + glog.V(0).Infof("retry processor shutting down, %d items remaining in queue", f.DeletionRetryQueue.Size()) + return + case <-ticker.C: + // Process all ready items in batches until queue is empty + totalProcessed := 0 + for { + readyItems := f.DeletionRetryQueue.GetReadyItems(DeletionRetryBatchSize) + if len(readyItems) == 0 { + break } + + f.processRetryBatch(readyItems, lookupFunc) + totalProcessed += len(readyItems) } - }) - if deletionCount == 0 { - time.Sleep(1123 * time.Millisecond) + if totalProcessed > 0 { + glog.V(1).Infof("retried deletion of %d files", totalProcessed) + } } } } +// processRetryBatch attempts to retry deletion of files and processes results. +// Successfully deleted items are removed from tracking, retryable failures are +// re-queued with updated retry counts, and permanent errors are logged and discarded. +func (f *Filer) processRetryBatch(readyItems []*DeletionRetryItem, lookupFunc func([]string) (map[string]*operation.LookupResult, error)) { + // Extract file IDs from retry items + fileIds := make([]string, 0, len(readyItems)) + for _, item := range readyItems { + fileIds = append(fileIds, item.FileId) + } + + // Delete files and classify outcomes + outcomes := deleteFilesAndClassify(f.GrpcDialOption, fileIds, lookupFunc) + + // Process outcomes - iterate over readyItems to ensure all items are accounted for + var successCount, notFoundCount, retryCount, permanentErrorCount int + for _, item := range readyItems { + outcome := outcomes[item.FileId] + + switch outcome.status { + case deletionOutcomeSuccess: + successCount++ + f.DeletionRetryQueue.Remove(item) // Remove from queue (success) + glog.V(2).Infof("retry successful for %s after %d attempts", item.FileId, item.RetryCount) + case deletionOutcomeNotFound: + notFoundCount++ + f.DeletionRetryQueue.Remove(item) // Remove from queue (already deleted) + case deletionOutcomeRetryable, deletionOutcomeNoResult: + retryCount++ + if outcome.status == deletionOutcomeNoResult { + glog.Warningf("no deletion result for retried file %s, re-queuing to avoid loss", item.FileId) + } + f.DeletionRetryQueue.RequeueForRetry(item, outcome.errorMsg) + case deletionOutcomePermanent: + permanentErrorCount++ + f.DeletionRetryQueue.Remove(item) // Remove from queue (permanent failure) + glog.Warningf("permanent error on retry for %s after %d attempts: %s", item.FileId, item.RetryCount, outcome.errorMsg) + } + } + + if successCount > 0 || notFoundCount > 0 { + glog.V(1).Infof("retry: deleted %d files successfully, %d already deleted", successCount, notFoundCount) + } + if retryCount > 0 { + glog.V(1).Infof("retry: %d files still failing, will retry again later", retryCount) + } + if permanentErrorCount > 0 { + glog.Warningf("retry: %d files failed with permanent errors", permanentErrorCount) + } +} + func (f *Filer) DeleteUncommittedChunks(ctx context.Context, chunks []*filer_pb.FileChunk) { f.doDeleteChunks(ctx, chunks) } diff --git a/weed/filer/filer_deletion_test.go b/weed/filer/filer_deletion_test.go new file mode 100644 index 000000000..77ac2310f --- /dev/null +++ b/weed/filer/filer_deletion_test.go @@ -0,0 +1,308 @@ +package filer + +import ( + "container/heap" + "testing" + "time" +) + +func TestDeletionRetryQueue_AddAndRetrieve(t *testing.T) { + queue := NewDeletionRetryQueue() + + // Add items + queue.AddOrUpdate("file1", "is read only") + queue.AddOrUpdate("file2", "connection reset") + + if queue.Size() != 2 { + t.Errorf("Expected queue size 2, got %d", queue.Size()) + } + + // Items not ready yet (initial delay is 5 minutes) + readyItems := queue.GetReadyItems(10) + if len(readyItems) != 0 { + t.Errorf("Expected 0 ready items, got %d", len(readyItems)) + } + + // Size should remain unchanged + if queue.Size() != 2 { + t.Errorf("Expected queue size 2 after checking ready items, got %d", queue.Size()) + } +} + +func TestDeletionRetryQueue_ExponentialBackoff(t *testing.T) { + queue := NewDeletionRetryQueue() + + // Create an item + item := &DeletionRetryItem{ + FileId: "test-file", + RetryCount: 0, + NextRetryAt: time.Now(), + LastError: "test error", + } + + // Requeue multiple times to test backoff + delays := []time.Duration{} + + for i := 0; i < 5; i++ { + beforeTime := time.Now() + queue.RequeueForRetry(item, "error") + + // Calculate expected delay for this retry count + expectedDelay := InitialRetryDelay * time.Duration(1< MaxRetryDelay { + expectedDelay = MaxRetryDelay + } + + // Verify NextRetryAt is approximately correct + actualDelay := item.NextRetryAt.Sub(beforeTime) + delays = append(delays, actualDelay) + + // Allow small timing variance + timeDiff := actualDelay - expectedDelay + if timeDiff < 0 { + timeDiff = -timeDiff + } + if timeDiff > 100*time.Millisecond { + t.Errorf("Retry %d: expected delay ~%v, got %v (diff: %v)", i+1, expectedDelay, actualDelay, timeDiff) + } + + // Verify retry count incremented + if item.RetryCount != i+1 { + t.Errorf("Expected RetryCount %d, got %d", i+1, item.RetryCount) + } + + // Reset the heap for the next isolated test iteration + queue.lock.Lock() + queue.heap = retryHeap{} + queue.lock.Unlock() + } + + t.Logf("Exponential backoff delays: %v", delays) +} + +func TestDeletionRetryQueue_OverflowProtection(t *testing.T) { + queue := NewDeletionRetryQueue() + + // Create an item with very high retry count + item := &DeletionRetryItem{ + FileId: "test-file", + RetryCount: 60, // High count that would cause overflow without protection + NextRetryAt: time.Now(), + LastError: "test error", + } + + // Should not panic and should cap at MaxRetryDelay + queue.RequeueForRetry(item, "error") + + delay := time.Until(item.NextRetryAt) + if delay > MaxRetryDelay+time.Second { + t.Errorf("Delay exceeded MaxRetryDelay: %v > %v", delay, MaxRetryDelay) + } +} + +func TestDeletionRetryQueue_MaxAttemptsReached(t *testing.T) { + queue := NewDeletionRetryQueue() + + // Add item + queue.AddOrUpdate("file1", "error") + + // Manually set retry count to max + queue.lock.Lock() + item, exists := queue.itemIndex["file1"] + if !exists { + queue.lock.Unlock() + t.Fatal("Item not found in queue") + } + item.RetryCount = MaxRetryAttempts + item.NextRetryAt = time.Now().Add(-1 * time.Second) // Ready now + heap.Fix(&queue.heap, item.heapIndex) + queue.lock.Unlock() + + // Try to get ready items - should be returned for the last retry (attempt #10) + readyItems := queue.GetReadyItems(10) + if len(readyItems) != 1 { + t.Fatalf("Expected 1 item for last retry, got %d", len(readyItems)) + } + + // Requeue it, which will increment its retry count beyond the max + queue.RequeueForRetry(readyItems[0], "final error") + + // Manually make it ready again + queue.lock.Lock() + item, exists = queue.itemIndex["file1"] + if !exists { + queue.lock.Unlock() + t.Fatal("Item not found in queue after requeue") + } + item.NextRetryAt = time.Now().Add(-1 * time.Second) + heap.Fix(&queue.heap, item.heapIndex) + queue.lock.Unlock() + + // Now it should be discarded (retry count is 11, exceeds max of 10) + readyItems = queue.GetReadyItems(10) + if len(readyItems) != 0 { + t.Errorf("Expected 0 items (max attempts exceeded), got %d", len(readyItems)) + } + + // Should be removed from queue + if queue.Size() != 0 { + t.Errorf("Expected queue size 0 after max attempts exceeded, got %d", queue.Size()) + } +} + +func TestCalculateBackoff(t *testing.T) { + testCases := []struct { + retryCount int + expectedDelay time.Duration + description string + }{ + {1, InitialRetryDelay, "first retry"}, + {2, InitialRetryDelay * 2, "second retry"}, + {3, InitialRetryDelay * 4, "third retry"}, + {4, InitialRetryDelay * 8, "fourth retry"}, + {5, InitialRetryDelay * 16, "fifth retry"}, + {10, MaxRetryDelay, "capped at max delay"}, + {65, MaxRetryDelay, "overflow protection (shift > 63)"}, + {100, MaxRetryDelay, "very high retry count"}, + } + + for _, tc := range testCases { + result := calculateBackoff(tc.retryCount) + if result != tc.expectedDelay { + t.Errorf("%s (retry %d): expected %v, got %v", + tc.description, tc.retryCount, tc.expectedDelay, result) + } + } +} + +func TestIsRetryableError(t *testing.T) { + testCases := []struct { + error string + retryable bool + description string + }{ + {"volume 123 is read only", true, "read-only volume"}, + {"connection reset by peer", true, "connection reset"}, + {"timeout exceeded", true, "timeout"}, + {"deadline exceeded", true, "deadline exceeded"}, + {"context canceled", true, "context canceled"}, + {"lookup error: volume not found", true, "lookup error"}, + {"connection refused", true, "connection refused"}, + {"too many requests", true, "rate limiting"}, + {"service unavailable", true, "service unavailable"}, + {"i/o timeout", true, "I/O timeout"}, + {"broken pipe", true, "broken pipe"}, + {"not found", false, "not found (not retryable)"}, + {"invalid file id", false, "invalid input (not retryable)"}, + {"", false, "empty error"}, + } + + for _, tc := range testCases { + result := isRetryableError(tc.error) + if result != tc.retryable { + t.Errorf("%s: expected retryable=%v, got %v for error: %q", + tc.description, tc.retryable, result, tc.error) + } + } +} + +func TestDeletionRetryQueue_HeapOrdering(t *testing.T) { + queue := NewDeletionRetryQueue() + + now := time.Now() + + // Add items with different retry times (out of order) + items := []*DeletionRetryItem{ + {FileId: "file3", RetryCount: 1, NextRetryAt: now.Add(30 * time.Second), LastError: "error3"}, + {FileId: "file1", RetryCount: 1, NextRetryAt: now.Add(10 * time.Second), LastError: "error1"}, + {FileId: "file2", RetryCount: 1, NextRetryAt: now.Add(20 * time.Second), LastError: "error2"}, + } + + // Add items directly (simulating internal state) + for _, item := range items { + queue.lock.Lock() + queue.itemIndex[item.FileId] = item + queue.heap = append(queue.heap, item) + queue.lock.Unlock() + } + + // Use container/heap.Init to establish heap property + queue.lock.Lock() + heap.Init(&queue.heap) + queue.lock.Unlock() + + // Verify heap maintains min-heap property (earliest time at top) + queue.lock.Lock() + if queue.heap[0].FileId != "file1" { + t.Errorf("Expected file1 at heap top (earliest time), got %s", queue.heap[0].FileId) + } + queue.lock.Unlock() + + // Set all items to ready while preserving their relative order + queue.lock.Lock() + for _, item := range queue.itemIndex { + // Shift all times back by 40 seconds to make them ready, but preserve order + item.NextRetryAt = item.NextRetryAt.Add(-40 * time.Second) + } + heap.Init(&queue.heap) // Re-establish heap property after modification + queue.lock.Unlock() + + // GetReadyItems should return in NextRetryAt order + readyItems := queue.GetReadyItems(10) + expectedOrder := []string{"file1", "file2", "file3"} + + if len(readyItems) != 3 { + t.Fatalf("Expected 3 ready items, got %d", len(readyItems)) + } + + for i, item := range readyItems { + if item.FileId != expectedOrder[i] { + t.Errorf("Item %d: expected %s, got %s", i, expectedOrder[i], item.FileId) + } + } +} + +func TestDeletionRetryQueue_DuplicateFileIds(t *testing.T) { + queue := NewDeletionRetryQueue() + + // Add same file ID twice with retryable error - simulates duplicate in batch + queue.AddOrUpdate("file1", "timeout error") + + // Verify only one item exists in queue + if queue.Size() != 1 { + t.Fatalf("Expected queue size 1 after first add, got %d", queue.Size()) + } + + // Get initial retry count + queue.lock.Lock() + item1, exists := queue.itemIndex["file1"] + if !exists { + queue.lock.Unlock() + t.Fatal("Item not found in queue after first add") + } + initialRetryCount := item1.RetryCount + queue.lock.Unlock() + + // Add same file ID again - should NOT increment retry count (just update error) + queue.AddOrUpdate("file1", "timeout error again") + + // Verify still only one item exists in queue (not duplicated) + if queue.Size() != 1 { + t.Errorf("Expected queue size 1 after duplicate add, got %d (duplicates detected)", queue.Size()) + } + + // Verify retry count did NOT increment (AddOrUpdate only updates error, not count) + queue.lock.Lock() + item2, exists := queue.itemIndex["file1"] + queue.lock.Unlock() + + if !exists { + t.Fatal("Item not found in queue after second add") + } + if item2.RetryCount != initialRetryCount { + t.Errorf("Expected RetryCount to stay at %d after duplicate add (should not increment), got %d", initialRetryCount, item2.RetryCount) + } + if item2.LastError != "timeout error again" { + t.Errorf("Expected LastError to be updated to 'timeout error again', got %q", item2.LastError) + } +}