fix: Fix handling of duplicate delete requests (#18460)

Author: sandeepsukhani

pkg/compactor/deletion/delete_request_batch.go

@@ -70,15 +70,15 @@ func (b *deleteRequestBatch) addDeleteRequest(dr *DeleteRequest) {
 	b.count++
 }
 
-func (b *deleteRequestBatch) checkDuplicate(deleteRequest DeleteRequest) error {
+func (b *deleteRequestBatch) checkDuplicate(deleteRequest DeleteRequest) (bool, error) {
 	ur, ok := b.deleteRequestsToProcess[deleteRequest.UserID]
 	if !ok {
-		return nil
+		return false, nil
 	}
 	for _, requestLoadedForProcessing := range ur.requests {
 		isDuplicate, err := requestLoadedForProcessing.IsDuplicate(&deleteRequest)
 		if err != nil {
-			return err
+			return false, err
 		}
 		if isDuplicate {
 			level.Info(util_log.Logger).Log(
@@ -88,10 +88,11 @@ func (b *deleteRequestBatch) checkDuplicate(deleteRequest DeleteRequest) error {
 				"user", deleteRequest.UserID,
 			)
 			b.duplicateRequests = append(b.duplicateRequests, deleteRequest)
+			return true, nil
 		}
 	}
 
-	return nil
+	return false, nil
 }
 
 func (b *deleteRequestBatch) expired(userID []byte, chk retention.Chunk, lbls labels.Labels, skipRequest func(*DeleteRequest) bool) (bool, filter.Func) {
@@ -163,3 +164,14 @@ func (b *deleteRequestBatch) getAllRequestsForUser(userID string) []*DeleteReque
 
 	return userRequests.requests
 }
+
+func (b *deleteRequestBatch) getAllRequests() []*DeleteRequest {
+	requests := make([]*DeleteRequest, 0, b.count)
+	for _, ur := range b.deleteRequestsToProcess {
+		for _, request := range ur.requests {
+			requests = append(requests, request)
+		}
+	}
+
+	return requests
+}

pkg/compactor/deletion/delete_requests_db_sqlite.go

@@ -30,8 +30,9 @@ const (
 )
 
 type sqlQuery struct {
-	query    string
-	execOpts *sqlitex.ExecOptions
+	query                  string
+	execOpts               *sqlitex.ExecOptions
+	postUpdateExecCallback func(numChanges int) error
 }
 
 type sqliteDB struct {
@@ -141,6 +142,11 @@ func (s *sqliteDB) Exec(ctx context.Context, updatesData bool, queries ...sqlQue
 		if err := sqlitex.Execute(conn, query.query, query.execOpts); err != nil {
 			return err
 		}
+		if updatesData && query.postUpdateExecCallback != nil {
+			if err := query.postUpdateExecCallback(conn.Changes()); err != nil {
+				return err
+			}
+		}
 	}
 
 	if updatesData {

pkg/compactor/deletion/delete_requests_manager.go

@@ -353,9 +353,13 @@ func (d *DeleteRequestsManager) loadDeleteRequestsToProcess(kind DeleteRequestsK
 				continue
 			}
 		}
-		if err := batch.checkDuplicate(deleteRequest); err != nil {
+		isDuplicate, err := batch.checkDuplicate(deleteRequest)
+		if err != nil {
 			return nil, err
 		}
+		if isDuplicate {
+			continue
+		}
 		if reqCount >= d.batchSize {
 			logBatchTruncation(reqCount, len(deleteRequests))
 			break