feat(scheduler): replace clone concurrency limit with cost-based admission

Replace the binary isCloneJob/MaxCloneConcurrency mechanism with a
generic cost model. Strategies now declare the cost of each job at
submit time, and the scheduler tracks total active cost against a
configurable budget (max-cost).

Cost constants defined in the git strategy:
  clone=4, snapshot=3, repack=2, fetch=1

Default max-cost is Concurrency * 4. A job is always admitted when
nothing else is running, even if its cost exceeds max-cost, to prevent
permanent starvation from misconfiguration.

This removes isCloneJob and the scheduler's knowledge of git-specific
job types. The scheduler now sees only costs and the strategy decides
what each job is worth.

Amp-Thread-ID: https://ampcode.com/threads/T-019d404e-21ec-723a-b211-c619925dd12e
Co-authored-by: Amp <amp@ampcode.com>

Author: worstell

internal/jobscheduler/jobs.go

@@ -16,14 +16,15 @@ import (
 )
 
 type Config struct {
-	Concurrency         int    `hcl:"concurrency" help:"The maximum number of concurrent jobs to run (0 means number of cores)." default:"4"`
-	MaxCloneConcurrency int    `hcl:"max-clone-concurrency" help:"Maximum number of concurrent clone jobs. Remaining worker slots are reserved for fetch/repack/snapshot jobs. 0 means no limit." default:"0"`
-	SchedulerDB         string `hcl:"scheduler-db" help:"Path to the scheduler state database." default:"${CACHEW_STATE}/scheduler.db"`
+	Concurrency int    `hcl:"concurrency" help:"The maximum number of concurrent jobs to run (0 means number of cores)." default:"4"`
+	MaxCost     int    `hcl:"max-cost" help:"Maximum total cost of concurrently running jobs. Each job declares its own cost at submission. 0 means Concurrency * 4." default:"0"`
+	SchedulerDB string `hcl:"scheduler-db" help:"Path to the scheduler state database." default:"${CACHEW_STATE}/scheduler.db"`
 }
 
 type queueJob struct {
 	id    string
 	queue string
+	cost  int
 	run   func(ctx context.Context) error
 }
 
@@ -43,27 +44,31 @@ type Scheduler interface {
 	//
 	// This is useful to avoid collisions across strategies.
 	WithQueuePrefix(prefix string) Scheduler
-	// Submit a job to the queue.
+	// Submit a job to the queue with a given cost.
 	//
 	// Jobs run concurrently across queues, but never within a queue.
-	Submit(queue, id string, run func(ctx context.Context) error)
+	// The cost is used for admission control: a job is only started when
+	// activeCost + cost <= maxCost.
+	Submit(queue, id string, cost int, run func(ctx context.Context) error)
 	// SubmitPeriodicJob submits a job to the queue that runs immediately, and then periodically after the interval.
 	//
 	// Jobs run concurrently across queues, but never within a queue.
-	SubmitPeriodicJob(queue, id string, interval time.Duration, run func(ctx context.Context) error)
+	// The cost is used for admission control: a job is only started when
+	// activeCost + cost <= maxCost.
+	SubmitPeriodicJob(queue, id string, cost int, interval time.Duration, run func(ctx context.Context) error)
 }
 
 type prefixedScheduler struct {
 	prefix    string
 	scheduler Scheduler
 }
 
-func (p *prefixedScheduler) Submit(queue, id string, run func(ctx context.Context) error) {
-	p.scheduler.Submit(queue, p.prefix+id, run)
+func (p *prefixedScheduler) Submit(queue, id string, cost int, run func(ctx context.Context) error) {
+	p.scheduler.Submit(queue, p.prefix+id, cost, run)
 }
 
-func (p *prefixedScheduler) SubmitPeriodicJob(queue, id string, interval time.Duration, run func(ctx context.Context) error) {
-	p.scheduler.SubmitPeriodicJob(queue, p.prefix+id, interval, run)
+func (p *prefixedScheduler) SubmitPeriodicJob(queue, id string, cost int, interval time.Duration, run func(ctx context.Context) error) {
+	p.scheduler.SubmitPeriodicJob(queue, p.prefix+id, cost, interval, run)
 }
 
 func (p *prefixedScheduler) WithQueuePrefix(prefix string) Scheduler {
@@ -74,15 +79,16 @@ func (p *prefixedScheduler) WithQueuePrefix(prefix string) Scheduler {
 }
 
 type RootScheduler struct {
-	workAvailable       chan bool
-	lock                sync.Mutex
-	queue               []queueJob
-	active              map[string]string // queue -> job id
-	activeClones        int
-	maxCloneConcurrency int
-	cancel              context.CancelFunc
-	store               ScheduleStore
-	metrics             *schedulerMetrics
+	workAvailable chan bool
+	lock          sync.Mutex
+	queue         []queueJob
+	active        map[string]string // queue -> job id
+	activeCost    int
+	activeCosts   map[string]int // queue -> cost of running job
+	maxCost       int
+	cancel        context.CancelFunc
+	store         ScheduleStore
+	metrics       *schedulerMetrics
 }
 
 var _ Scheduler = &RootScheduler{}
@@ -109,21 +115,21 @@ func New(ctx context.Context, config Config) (*RootScheduler, error) {
 			return nil, errors.Wrap(err, "create schedule store")
 		}
 	}
-	maxClones := config.MaxCloneConcurrency
-	if maxClones == 0 && config.Concurrency > 1 {
-		// Default: reserve at least half the workers for non-clone jobs.
-		maxClones = max(1, config.Concurrency/2)
+	maxCost := config.MaxCost
+	if maxCost == 0 {
+		maxCost = config.Concurrency * 4
 	}
 	m, err := newSchedulerMetrics()
 	if err != nil {
 		return nil, errors.Wrap(err, "create scheduler metrics")
 	}
 	q := &RootScheduler{
-		workAvailable:       make(chan bool, 1024),
-		active:              make(map[string]string),
-		maxCloneConcurrency: maxClones,
-		store:               store,
-		metrics:             m,
+		workAvailable: make(chan bool, 1024),
+		active:        make(map[string]string),
+		activeCosts:   make(map[string]int),
+		maxCost:       maxCost,
+		store:         store,
+		metrics:       m,
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	q.cancel = cancel
@@ -147,19 +153,19 @@ func (q *RootScheduler) WithQueuePrefix(prefix string) Scheduler {
 	}
 }
 
-func (q *RootScheduler) Submit(queue, id string, run func(ctx context.Context) error) {
+func (q *RootScheduler) Submit(queue, id string, cost int, run func(ctx context.Context) error) {
 	q.lock.Lock()
-	q.queue = append(q.queue, queueJob{queue: queue, id: id, run: run})
+	q.queue = append(q.queue, queueJob{queue: queue, id: id, cost: cost, run: run})
 	q.metrics.queueDepth.Record(context.Background(), int64(len(q.queue)))
 	q.lock.Unlock()
 	q.workAvailable <- true
 }
 
-func (q *RootScheduler) SubmitPeriodicJob(queue, id string, interval time.Duration, run func(ctx context.Context) error) {
+func (q *RootScheduler) SubmitPeriodicJob(queue, id string, cost int, interval time.Duration, run func(ctx context.Context) error) {
 	key := jobKey(queue, id)
 	delay := q.periodicDelay(key, interval)
 	submit := func() {
-		q.Submit(queue, id, func(ctx context.Context) error {
+		q.Submit(queue, id, cost, func(ctx context.Context) error {
 			err := run(ctx)
 			if q.store != nil {
 				if storeErr := q.store.SetLastRun(key, time.Now()); storeErr != nil {
@@ -168,7 +174,7 @@ func (q *RootScheduler) SubmitPeriodicJob(queue, id string, interval time.Durati
 			}
 			go func() {
 				time.Sleep(interval)
-				q.SubmitPeriodicJob(queue, id, interval, run)
+				q.SubmitPeriodicJob(queue, id, cost, interval, run)
 			}()
 			return errors.WithStack(err)
 		})
@@ -252,19 +258,12 @@ func jobType(id string) string {
 func (q *RootScheduler) markQueueInactive(queue string) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
-	if isCloneJob(q.active[queue]) {
-		q.activeClones--
-	}
+	q.activeCost -= q.activeCosts[queue]
+	delete(q.activeCosts, queue)
 	delete(q.active, queue)
 	q.recordGaugesLocked()
 }
 
-// isCloneJob returns true for job IDs that represent long-running clone operations
-// which should be subject to concurrency limits.
-func isCloneJob(id string) bool {
-	return strings.HasSuffix(id, "clone") || strings.HasSuffix(id, "deferred-mirror-restore")
-}
-
 // Take the next job for any queue that is not already running a job.
 func (q *RootScheduler) takeNextJob() (queueJob, bool) {
 	q.lock.Lock()
@@ -273,15 +272,14 @@ func (q *RootScheduler) takeNextJob() (queueJob, bool) {
 		if _, active := q.active[job.queue]; active {
 			continue
 		}
-		if q.maxCloneConcurrency > 0 && isCloneJob(job.id) && q.activeClones >= q.maxCloneConcurrency {
+		if q.activeCost > 0 && q.activeCost+job.cost > q.maxCost {
 			continue
 		}
 		q.queue = append(q.queue[:i], q.queue[i+1:]...)
 		q.workAvailable <- true
 		q.active[job.queue] = job.id
-		if isCloneJob(job.id) {
-			q.activeClones++
-		}
+		q.activeCost += job.cost
+		q.activeCosts[job.queue] = job.cost
 		q.recordGaugesLocked()
 		return job, true
 	}
@@ -293,5 +291,5 @@ func (q *RootScheduler) recordGaugesLocked() {
 	ctx := context.Background()
 	q.metrics.queueDepth.Record(ctx, int64(len(q.queue)))
 	q.metrics.activeWorkers.Record(ctx, int64(len(q.active)))
-	q.metrics.activeClones.Record(ctx, int64(q.activeClones))
+	q.metrics.activeCost.Record(ctx, int64(q.activeCost))
 }