diff --git a/weed/filer/filer_deletion.go b/weed/filer/filer_deletion.go index 4e322d79b5..1ee66eba97 100644 --- a/weed/filer/filer_deletion.go +++ b/weed/filer/filer_deletion.go @@ -21,12 +21,17 @@ import ( ) const ( - // Maximum number of retry attempts for failed deletions + // Backed-off retry attempts before a failed deletion moves to the dead letter queues MaxRetryAttempts = 10 // Initial retry delay (will be doubled with each attempt) InitialRetryDelay = 5 * time.Minute // Maximum retry delay MaxRetryDelay = 6 * time.Hour + // Deletions still failing after MaxRetryAttempts move to a dead letter + // queue retried after DeadLetterRetryDelay, then to a second one retried + // after DeadLetterFinalRetryDelay, and are dropped if that fails too. + DeadLetterRetryDelay = 24 * time.Hour + DeadLetterFinalRetryDelay = 48 * time.Hour // Interval for checking retry queue for ready items DeletionRetryPollInterval = 1 * time.Minute // Maximum number of items to process per retry iteration @@ -163,6 +168,31 @@ func calculateBackoff(retryCount int) time.Duration { return delay } +const deadLetterQueues = 2 + +// deadLetterQueue reports which dead letter queue an item with this retry +// count is in: 0 while it is still in the regular retry schedule. +func deadLetterQueue(retryCount int) int { + if retryCount <= MaxRetryAttempts { + return 0 + } + return retryCount - MaxRetryAttempts +} + +// retryDelay is the wait before the attempt numbered retryCount. Items past +// the last dead letter queue are due at once so the next poll drops them. +func retryDelay(retryCount int) time.Duration { + switch deadLetterQueue(retryCount) { + case 0: + return calculateBackoff(retryCount) + case 1: + return DeadLetterRetryDelay + case 2: + return DeadLetterFinalRetryDelay + } + return 0 +} + // 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, ledgerEpoch uint64) { @@ -216,8 +246,7 @@ func (q *DeletionRetryQueue) RequeueForRetry(item *DeletionRetryItem, errorMsg s item.RetryCount++ item.LastError = errorMsg - // Calculate next retry time with exponential backoff - delay := calculateBackoff(item.RetryCount) + delay := retryDelay(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) @@ -249,14 +278,13 @@ func (q *DeletionRetryQueue) GetReadyItems(maxItems int) (ready []*DeletionRetry // Remove from heap but keep in itemIndex with inFlight flag heap.Pop(&q.heap) - if item.RetryCount <= MaxRetryAttempts { + if deadLetterQueue(item.RetryCount) <= deadLetterQueues { item.inFlight = true // Mark as being processed ready = append(ready, item) } else { - // Max attempts reached, log and discard completely delete(q.itemIndex, item.FileId) expired = append(expired, item) - glog.Warningf("max retry attempts (%d) reached for %s, last error: %s", MaxRetryAttempts, item.FileId, item.LastError) + glog.Warningf("giving up deleting %s after %d retries, last error: %s", item.FileId, item.RetryCount-1, item.LastError) } } @@ -587,6 +615,7 @@ func (f *Filer) processRetryBatch(readyItems []*DeletionRetryItem, lookupFunc fu // Process outcomes - iterate over readyItems to ensure all items are accounted for var successCount, notFoundCount, retryCount, permanentErrorCount int + var deadLettered, finalDeadLettered int for _, item := range readyItems { outcome := outcomes[item.FileId] @@ -606,6 +635,12 @@ func (f *Filer) processRetryBatch(readyItems []*DeletionRetryItem, lookupFunc fu glog.Warningf("no deletion result for retried file %s, re-queuing to avoid loss", item.FileId) } // Keep in the durable ledger: still not confirmed. + switch deadLetterQueue(item.RetryCount + 1) { + case 1: + deadLettered++ + case 2: + finalDeadLettered++ + } f.DeletionRetryQueue.RequeueForRetry(item, outcome.errorMsg) case deletionOutcomePermanent: permanentErrorCount++ @@ -625,6 +660,12 @@ func (f *Filer) processRetryBatch(readyItems []*DeletionRetryItem, lookupFunc fu if retryCount > 0 { glog.V(1).Infof("retry: %d files still failing, will retry again later", retryCount) } + if deadLettered > 0 { + glog.Warningf("retry: %d files moved to the dead letter queue, next retry in %v", deadLettered, DeadLetterRetryDelay) + } + if finalDeadLettered > 0 { + glog.Warningf("retry: %d files moved to the second dead letter queue, next retry in %v", finalDeadLettered, DeadLetterFinalRetryDelay) + } if permanentErrorCount > 0 { glog.Warningf("retry: %d files failed with permanent errors", permanentErrorCount) } diff --git a/weed/filer/filer_deletion_test.go b/weed/filer/filer_deletion_test.go index c2a0cda5dc..3f1b921aea 100644 --- a/weed/filer/filer_deletion_test.go +++ b/weed/filer/filer_deletion_test.go @@ -102,49 +102,41 @@ func TestDeletionRetryQueue_OverflowProtection(t *testing.T) { func TestDeletionRetryQueue_MaxAttemptsReached(t *testing.T) { queue := NewDeletionRetryQueue() - - // Add item queue.AddOrUpdate("file1", "error", 0) - // Manually set retry count to max + makeReady := func() { + queue.lock.Lock() + defer queue.lock.Unlock() + item, exists := queue.itemIndex["file1"] + if !exists { + t.Fatal("Item not found in queue") + } + item.NextRetryAt = time.Now().Add(-1 * time.Second) + heap.Fix(&queue.heap, item.heapIndex) + } + 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.itemIndex["file1"].RetryCount = MaxRetryAttempts queue.lock.Unlock() + makeReady() - // 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)) + // The last backed-off retry, then one attempt from each dead letter queue. + for _, wantDelay := range []time.Duration{DeadLetterRetryDelay, DeadLetterFinalRetryDelay, 0} { + readyItems, expired := queue.GetReadyItems(10) + if len(readyItems) != 1 || len(expired) != 0 { + t.Fatalf("Expected 1 ready item, got %d ready and %d expired", len(readyItems), len(expired)) + } + queue.RequeueForRetry(readyItems[0], "still offline") + if delay := time.Until(readyItems[0].NextRetryAt); delay > wantDelay || delay < wantDelay-time.Minute { + t.Fatalf("attempt %d: expected next retry in ~%v, got %v", readyItems[0].RetryCount, wantDelay, delay) + } + makeReady() } - // 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") + readyItems, expired := queue.GetReadyItems(10) + if len(readyItems) != 0 || len(expired) != 1 { + t.Errorf("Expected the item to be dropped after the dead letter queues, got %d ready and %d expired", len(readyItems), len(expired)) } - 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()) }