Set limit on pulp orphan cleanup tasks

Set global cap and add new constant:
maxGetTaskAttempts (default is 5).
HMS-10580 API - pulp orphan cleanup runs too many tasks

Author: swadeley

pkg/clients/pulp_client/task.go

@@ -23,6 +23,10 @@ const (
 	FAILED    string = "failed"
 )
 
+// maxGetTaskAttempts is how many consecutive failed GetTask calls PollTask tolerates before
+// returning an error from a poll iteration (with SleepWithBackoff between attempts).
+const maxGetTaskAttempts = 5
+
 // GetTask Fetch a pulp task
 func (r pulpDaoImpl) GetTask(ctx context.Context, taskHref string) (zest.TaskResponse, error) {
 	ctx, client, err := getZestClient(ctx)
@@ -67,14 +71,23 @@ func (r pulpDaoImpl) CancelTask(ctx context.Context, taskHref string) (zest.Task
 // PollTask Poll a task and return the final task object
 func (r pulpDaoImpl) PollTask(ctx context.Context, taskHref string) (*zest.TaskResponse, error) {
 	var task zest.TaskResponse
-	var err error
 	inProgress := true
 	pollCount := 1
 	logger := zerolog.Ctx(ctx)
 	for inProgress {
-		task, err = r.GetTask(ctx, taskHref)
-		if err != nil {
-			return nil, err
+		var err error
+		for attempt := 1; attempt <= maxGetTaskAttempts; attempt++ {
+			task, err = r.GetTask(ctx, taskHref)
+			if err == nil {
+				break
+			}
+			if attempt == maxGetTaskAttempts {
+				return nil, err
+			}
+			logger.Debug().Err(err).Int("attempt", attempt).Str("task_href", taskHref).
+				Msg("GetTask failed while polling pulp task; retrying")
+			SleepWithBackoff(pollCount)
+			pollCount++
 		}
 		taskState := *task.State
 		switch {

pkg/external_repos/commands/cleanup.go

@@ -6,6 +6,7 @@ import (
 	"slices"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/content-services/content-sources-backend/pkg/clients/pulp_client"
@@ -116,7 +117,8 @@ func enqueueSnapshotsCleanup(ctx context.Context, olderThanDays int, batchSize i
 		return fmt.Errorf("error getting repository configurations: %v", err)
 	}
 	if batchSize > 0 {
-		repoConfigs = repoConfigs[:batchSize] // Limit to batch size
+		limit := min(batchSize, len(repoConfigs))
+		repoConfigs = repoConfigs[:limit]
 	}
 	log.Info().Msgf("Snapshot cleanup: processing %d repositories with outdated snapshots", len(repoConfigs))
 
@@ -143,12 +145,15 @@ func enqueueSnapshotCleanupForRepoConfig(ctx context.Context, taskClient client.
 	slices.SortFunc(snaps, func(s1, s2 models.Snapshot) int {
 		return s1.CreatedAt.Compare(s2.CreatedAt)
 	})
+
+	keepIndex := len(snaps) - 1
+	cutoffTime := time.Now().Add(-time.Duration(olderThanDays) * 24 * time.Hour)
 	toBeDeletedSnapUUIDs := make([]string, 0, len(snaps))
 	for i, snap := range snaps {
-		if i == len(snaps)-1 && len(toBeDeletedSnapUUIDs) == len(snaps)-1 {
+		if i == keepIndex && len(toBeDeletedSnapUUIDs) == len(snaps)-1 {
 			break
 		}
-		if snap.CreatedAt.Before(time.Now().Add(-time.Duration(olderThanDays) * 24 * time.Hour)) {
+		if snap.CreatedAt.Before(cutoffTime) {
 			toBeDeletedSnapUUIDs = append(toBeDeletedSnapUUIDs, snap.UUID)
 		}
 	}
@@ -194,15 +199,23 @@ func enqueueSnapshotCleanupForRepoConfig(ctx context.Context, taskClient client.
 }
 
 func pulpOrphanCleanup(ctx context.Context, db *gorm.DB, batchSize int) error {
-	var err error
 	daoReg := dao.GetDaoRegistry(db)
 
 	domains, err := daoReg.Domain.List(ctx)
 	if err != nil {
 		log.Panic().Err(err).Msg("orphan cleanup error: error listing orgs")
 	}
 
-	for i := 0; i < len(domains); i += batchSize {
+	return runPulpOrphanCleanupForDomains(ctx, domains, batchSize, pulp_client.GetPulpClientWithDomain)
+}
+
+// runPulpOrphanCleanupForDomains runs orphan cleanup in batches and stops if any PollTask fails
+// (PollTask already retries transient GetTask errors).
+func runPulpOrphanCleanupForDomains(ctx context.Context, domains []models.Domain, batchSize int, getPulpClient func(domainName string) pulp_client.PulpClient) error {
+	var abort atomic.Bool
+	var firstErr error
+
+	for i := 0; i < len(domains) && !abort.Load(); i += batchSize {
 		end := i + batchSize
 		if end > len(domains) {
 			end = len(domains)
@@ -218,24 +231,33 @@ func pulpOrphanCleanup(ctx context.Context, db *gorm.DB, batchSize int) error {
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
-
-				pulpClient := pulp_client.GetPulpClientWithDomain(domainName)
+				if abort.Load() {
+					return
+				}
+				pulpClient := getPulpClient(domainName)
 				cleanupTask, err := pulpClient.OrphanCleanup(ctx)
 				if err != nil {
 					logger.Error().Err(err).Msgf("error starting orphan cleanup")
 					return
 				}
 				logger.Info().Str("task_href", cleanupTask).Msgf("running orphan cleanup for org: %v", org)
 
-				_, err = pulp_client.GetGlobalPulpClient().PollTask(ctx, cleanupTask)
+				_, err = pulpClient.PollTask(ctx, cleanupTask)
 				if err != nil {
-					logger.Error().Err(err).Msgf("error polling pulp task for orphan cleanup")
+					logger.Error().Err(err).
+						Msg("error polling pulp task for orphan cleanup; remaining domains will be skipped")
+					if abort.CompareAndSwap(false, true) {
+						firstErr = err
+					}
 					return
 				}
 			}()
 		}
 		wg.Wait()
 	}
+	if abort.Load() {
+		return fmt.Errorf("pulp orphan cleanup aborted after failed task poll: %w", firstErr)
+	}
 	return nil
 }
 

pkg/external_repos/commands/cleanup_pulp_orphan_test.go

@@ -0,0 +1,58 @@
+package commands
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+
+	"github.com/content-services/content-sources-backend/pkg/clients/pulp_client"
+	"github.com/content-services/content-sources-backend/pkg/models"
+	"github.com/rs/zerolog"
+	zlog "github.com/rs/zerolog/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// When poll fails, further batches must not run. batchSize 3 runs three goroutines in parallel per batch.
+func TestRunPulpOrphanCleanupForDomains_StopsOnFirstPollFailure(t *testing.T) {
+	prev := zlog.Logger
+	zlog.Logger = zerolog.Nop()
+	t.Cleanup(func() { zlog.Logger = prev })
+
+	ctx := context.Background()
+	domains := []models.Domain{
+		{OrgId: "1", DomainName: "d1"},
+		{OrgId: "2", DomainName: "d2"},
+		{OrgId: "3", DomainName: "d3"},
+		{OrgId: "4", DomainName: "d4"},
+		{OrgId: "5", DomainName: "d5"},
+		{OrgId: "6", DomainName: "d6"},
+		{OrgId: "7", DomainName: "d7"},
+		{OrgId: "8", DomainName: "d8"},
+		{OrgId: "9", DomainName: "d9"},
+	}
+	const batchSize = 3
+	taskHref := "https://pulp.example/api/v3/tasks/fake-href/"
+
+	var orphanStarts atomic.Int32
+	mockPulp := pulp_client.NewMockPulpClient(t)
+	mockPulp.On("OrphanCleanup", mock.Anything).Return(taskHref, nil).Run(func(mock.Arguments) {
+		orphanStarts.Add(1)
+	})
+	mockPulp.On("PollTask", mock.Anything, taskHref).Return(nil, errors.New("injected poll failure"))
+
+	getPulpClient := func(_ string) pulp_client.PulpClient {
+		return mockPulp
+	}
+
+	err := runPulpOrphanCleanupForDomains(ctx, domains, batchSize, getPulpClient)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "pulp orphan cleanup aborted")
+	assert.Contains(t, err.Error(), "failed task poll")
+
+	starts := orphanStarts.Load()
+	assert.GreaterOrEqual(t, starts, int32(1), "at least one domain in the first batch should start orphan cleanup")
+	assert.LessOrEqual(t, starts, int32(batchSize), "only the first batch should run before abort (domains 4–9 must not be processed)")
+}