Add PULL_LIMIT and PUSH_LIMIT to cron.update_mirror task

Add `PULL_LIMIT` and `PUSH_LIMIT` to cron.update_mirror task to limit
the number of mirrors added to the queue each time the cron task is run.

Fix #16982

Signed-off-by: Andrew Thornton <art27@cantab.net>

custom/conf/app.example.ini

@@ -1705,6 +1705,10 @@ PATH =
 ;RUN_AT_START = false
 ;; Notice if not success
 ;NO_SUCCESS_NOTICE = true
+;; Limit the number of mirrors added to the queue to this number (set to -1 to add all)
+;PULL_LIMIT=50
+;; Limit the number of mirrors added to the queue to this number (set to -1 to add all)
+;PUSH_LIMIT=50
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

docs/content/doc/advanced/config-cheat-sheet.en-us.md

@@ -791,6 +791,8 @@ NB: You must have `DISABLE_ROUTER_LOG` set to `false` for this option to take ef
 
 - `SCHEDULE`: **@every 10m**: Cron syntax for scheduling update mirrors, e.g. `@every 3h`.
 - `NO_SUCCESS_NOTICE`: **true**: The cron task for update mirrors success report is not very useful - as it just means that the mirrors have been queued. Therefore this is turned off by default.
+- `PULL_LIMIT`: **50**: Limit the number of mirrors added to the queue to this number (set to -1 to add all).
+- `PUSH_LIMIT`: **50**: Limit the number of mirrors added to the queue to this number (set to -1 to add all).
 
 #### Cron - Repository Health Check (`cron.repo_health_check`)
 

models/repo_mirror.go

@@ -119,6 +119,7 @@ func MirrorsIterate(f func(idx int, bean interface{}) error) error {
 	return db.GetEngine(db.DefaultContext).
 		Where("next_update_unix<=?", time.Now().Unix()).
 		And("next_update_unix!=0").
+		OrderBy("updated_unix ASC").
 		Iterate(new(Mirror), f)
 }
 

models/repo_pushmirror.go

@@ -107,5 +107,6 @@ func PushMirrorsIterate(f func(idx int, bean interface{}) error) error {
 	return db.GetEngine(db.DefaultContext).
 		Where("last_update + (`interval` / ?) <= ?", time.Second, time.Now().Unix()).
 		And("`interval` != 0").
+		OrderBy("last_update ASC").
 		Iterate(new(PushMirror), f)
 }

modules/cron/tasks_basic.go

@@ -17,13 +17,24 @@ import (
 )
 
 func registerUpdateMirrorTask() {
-	RegisterTaskFatal("update_mirrors", &BaseConfig{
-		Enabled:         true,
-		RunAtStart:      false,
-		Schedule:        "@every 10m",
-		NoSuccessNotice: true,
-	}, func(ctx context.Context, _ *models.User, _ Config) error {
-		return mirror_service.Update(ctx)
+	type UpdateMirrorTaskConfig struct {
+		BaseConfig
+		PullLimit int
+		PushLimit int
+	}
+
+	RegisterTaskFatal("update_mirrors", &UpdateMirrorTaskConfig{
+		BaseConfig: BaseConfig{
+			Enabled:         true,
+			RunAtStart:      false,
+			Schedule:        "@every 10m",
+			NoSuccessNotice: true,
+		},
+		PullLimit: 50,
+		PushLimit: 50,
+	}, func(ctx context.Context, _ *models.User, cfg Config) error {
+		umtc := cfg.(*UpdateMirrorTaskConfig)
+		return mirror_service.Update(ctx, umtc.PullLimit, umtc.PushLimit)
 	})
 }
 

services/mirror/mirror.go

@@ -46,14 +46,18 @@ func doMirrorSync(ctx context.Context, req *SyncRequest) {
 }
 
 // Update checks and updates mirror repositories.
-func Update(ctx context.Context) error {
+func Update(ctx context.Context, pullLimit, pushLimit int) error {
 	if !setting.Mirror.Enabled {
 		log.Warn("Mirror feature disabled, but cron job enabled: skip update")
 		return nil
 	}
 	log.Trace("Doing: Update")
 
-	handler := func(idx int, bean interface{}) error {
+	requested := 0
+
+	doneErr := fmt.Errorf("reached limit")
+
+	handler := func(idx int, bean interface{}, limit int) error {
 		var item SyncRequest
 		if m, ok := bean.(*models.Mirror); ok {
 			if m.Repo == nil {
@@ -78,19 +82,50 @@ func Update(ctx context.Context) error {
 			return nil
 		}
 
+		// Check we've not been cancelled
 		select {
 		case <-ctx.Done():
-			return fmt.Errorf("Aborted")
+			return fmt.Errorf("aborted")
 		default:
-			return mirrorQueue.Push(&item)
 		}
+
+		// Check if this request is already in the queue
+		has, err := mirrorQueue.Has(&item)
+		if err != nil {
+			return err
+		}
+		if has {
+			return nil
+		}
+
+		// Double check we've not been cancelled
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("aborted")
+		default:
+		}
+
+		// Push to the Queue
+		if err := mirrorQueue.Push(&item); err != nil {
+			return err
+		}
+
+		requested++
+		if limit > 0 && requested > limit {
+			return doneErr
+		}
+		return nil
 	}
 
-	if err := models.MirrorsIterate(handler); err != nil {
+	if err := models.MirrorsIterate(func(idx int, bean interface{}) error {
+		return handler(idx, bean, pullLimit)
+	}); err != nil && err != doneErr {
 		log.Error("MirrorsIterate: %v", err)
 		return err
 	}
-	if err := models.PushMirrorsIterate(handler); err != nil {
+	if err := models.PushMirrorsIterate(func(idx int, bean interface{}) error {
+		return handler(idx, bean, pushLimit)
+	}); err != nil && err != doneErr {
 		log.Error("PushMirrorsIterate: %v", err)
 		return err
 	}