feat(scheduler): add crontab(5) random ~ syntax support (#5233)

* feat(scheduler): add CrontabSchedule with crontab(5) random ~ syntax Implement ParseCrontab() that extends robfig/cron with support for the crontab(5) random ~ operator (e.g., 0~30 * * * *). Random values are resolved fresh on each Next() call for load spreading. Supports A~B, ~B, A~, and bare ~ forms in all 6 fields (including seconds). Expressions without ~ delegate to robfig's standard parser with zero overhead. Integrates into scheduler.Add() and conf.validateSchedule() so that scanner.schedule and backup.schedule config values accept ~ syntax. * refactor(scheduler): resolve random ~ values once at parse time Change from per-Next() randomization to per-parse randomization, matching crontab(5) semantics. This prevents double-firing within the same period when random values land after the current time. ParseCrontab now resolves ~ fields to concrete values, substitutes them into the spec string, and delegates to robfig's parser. This eliminates CrontabSchedule, randomField, and resolveField entirely. * test(scheduler): replace WaitGroup with channel for job execution synchronization Signed-off-by: Deluan <deluan@navidrome.org> --------- Signed-off-by: Deluan <deluan@navidrome.org>
2026-03-20 08:57:13 -04:00
parent a4c289b28c
commit 5cd1fcb492
5 changed files with 347 additions and 33 deletions
@@ -0,0 +1,133 @@
+package scheduler
+
+import (
+	"fmt"
+	"math/rand/v2"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/robfig/cron/v3"
+)
+
+var parser = cron.NewParser(
+	cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
+)
+
+// ParseCrontab parses a cron expression with support for the crontab(5) random ~ syntax.
+// Random values are resolved once at parse time. If no ~ is present, it delegates to
+// robfig/cron's standard parser. Duration strings (e.g., "5m") are converted to "@every 5m".
+func ParseCrontab(spec string) (cron.Schedule, error) {
+	if spec == "" {
+		return nil, fmt.Errorf("empty spec string")
+	}
+
+	if _, err := time.ParseDuration(spec); err == nil {
+		spec = "@every " + spec
+	}
+
+	if !strings.Contains(spec, "~") {
+		return parser.Parse(spec)
+	}
+
+	// Handle TZ=/CRON_TZ= prefix
+	var tzPrefix string
+	if strings.HasPrefix(spec, "TZ=") || strings.HasPrefix(spec, "CRON_TZ=") {
+		i := strings.Index(spec, " ")
+		if i == -1 {
+			return nil, fmt.Errorf("missing spec after timezone")
+		}
+		tzPrefix = spec[:i] + " "
+		spec = strings.TrimSpace(spec[i:])
+	}
+
+	// @ descriptors cannot contain ~
+	if strings.HasPrefix(spec, "@") {
+		return nil, fmt.Errorf("random ~ syntax cannot be used with descriptors: %s", spec)
+	}
+
+	fields := strings.Fields(spec)
+	fields, err := normalizeFields(fields)
+	if err != nil {
+		return nil, err
+	}
+
+	// Resolve each ~ field to a concrete random value
+	for i, field := range fields {
+		if !strings.Contains(field, "~") {
+			continue
+		}
+		if strings.ContainsAny(field, ",/") {
+			return nil, fmt.Errorf("random ~ cannot be combined with lists or steps: %s", field)
+		}
+		v, parseErr := resolveRandomField(field, fieldBounds[i])
+		if parseErr != nil {
+			return nil, parseErr
+		}
+		fields[i] = strconv.FormatUint(uint64(v), 10)
+	}
+
+	// Re-assemble and parse with robfig
+	resolved := tzPrefix + strings.Join(fields, " ")
+	return parser.Parse(resolved)
+}
+
+type bounds struct {
+	min, max uint
+}
+
+var fieldBounds = [6]bounds{
+	{0, 59}, // Second
+	{0, 59}, // Minute
+	{0, 23}, // Hour
+	{1, 31}, // Dom
+	{1, 12}, // Month
+	{0, 6},  // Dow
+}
+
+// resolveRandomField parses a ~ field and returns a random value within the range.
+func resolveRandomField(field string, b bounds) (uint, error) {
+	parts := strings.SplitN(field, "~", 2)
+
+	min := b.min
+	max := b.max
+
+	if parts[0] != "" {
+		v, err := strconv.ParseUint(parts[0], 10, 0)
+		if err != nil {
+			return 0, fmt.Errorf("invalid random range start: %s", parts[0])
+		}
+		min = uint(v)
+	}
+
+	if parts[1] != "" {
+		v, err := strconv.ParseUint(parts[1], 10, 0)
+		if err != nil {
+			return 0, fmt.Errorf("invalid random range end: %s", parts[1])
+		}
+		max = uint(v)
+	}
+
+	if min < b.min {
+		return 0, fmt.Errorf("random range start (%d) below minimum (%d): %s", min, b.min, field)
+	}
+	if max > b.max {
+		return 0, fmt.Errorf("random range end (%d) above maximum (%d): %s", max, b.max, field)
+	}
+	if min > max {
+		return 0, fmt.Errorf("random range start (%d) beyond end (%d): %s", min, max, field)
+	}
+
+	return min + uint(rand.IntN(int(max-min+1))), nil //nolint:gosec // Cryptographic randomness not needed for schedule jitter
+}
+
+func normalizeFields(fields []string) ([]string, error) {
+	switch len(fields) {
+	case 5:
+		return append([]string{"0"}, fields...), nil
+	case 6:
+		return fields, nil
+	default:
+		return nil, fmt.Errorf("expected 5 or 6 fields, found %d: %v", len(fields), fields)
+	}
+}
@@ -0,0 +1,194 @@
+package scheduler
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/robfig/cron/v3"
+)
+
+var _ = Describe("ParseCrontab", func() {
+	Describe("standard expressions", func() {
+		It("parses a 5-field expression", func() {
+			sched, err := ParseCrontab("5 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sched).To(BeAssignableToTypeOf(&cron.SpecSchedule{}))
+		})
+
+		It("parses a 6-field expression with seconds", func() {
+			sched, err := ParseCrontab("30 5 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sched).To(BeAssignableToTypeOf(&cron.SpecSchedule{}))
+		})
+
+		It("converts duration string to @every", func() {
+			sched, err := ParseCrontab("5m")
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sched).To(BeAssignableToTypeOf(cron.ConstantDelaySchedule{}))
+		})
+
+		It("returns error for empty string", func() {
+			_, err := ParseCrontab("")
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("random ~ syntax", func() {
+		It("resolves A~B to a value within range", func() {
+			sched, err := ParseCrontab("0~30 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			minute := findSetBit(spec.Minute)
+			Expect(minute).To(BeNumerically(">=", 0))
+			Expect(minute).To(BeNumerically("<=", 30))
+		})
+
+		It("resolves ~ alone to full field range", func() {
+			sched, err := ParseCrontab("~ * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			minute := findSetBit(spec.Minute)
+			Expect(minute).To(BeNumerically(">=", 0))
+			Expect(minute).To(BeNumerically("<=", 59))
+		})
+
+		It("resolves ~B as min~B", func() {
+			sched, err := ParseCrontab("~15 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			minute := findSetBit(spec.Minute)
+			Expect(minute).To(BeNumerically(">=", 0))
+			Expect(minute).To(BeNumerically("<=", 15))
+		})
+
+		It("resolves A~ as A~max", func() {
+			sched, err := ParseCrontab("15~ * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			minute := findSetBit(spec.Minute)
+			Expect(minute).To(BeNumerically(">=", 15))
+			Expect(minute).To(BeNumerically("<=", 59))
+		})
+
+		It("resolves multiple random fields independently", func() {
+			sched, err := ParseCrontab("0~30 0~12 * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			Expect(findSetBit(spec.Minute)).To(BeNumerically("<=", 30))
+			Expect(findSetBit(spec.Hour)).To(BeNumerically("<=", 12))
+		})
+
+		It("resolves ~ in DOM field with correct bounds", func() {
+			sched, err := ParseCrontab("0 0 ~ * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			dom := findSetBit(spec.Dom)
+			Expect(dom).To(BeNumerically(">=", 1))
+			Expect(dom).To(BeNumerically("<=", 31))
+		})
+
+		It("resolves ~ in month field with correct bounds", func() {
+			sched, err := ParseCrontab("0 0 1 ~ *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			month := findSetBit(spec.Month)
+			Expect(month).To(BeNumerically(">=", 1))
+			Expect(month).To(BeNumerically("<=", 12))
+		})
+
+		It("resolves ~ in DOW field with correct bounds", func() {
+			sched, err := ParseCrontab("0 0 * * ~")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			dow := findSetBit(spec.Dow)
+			Expect(dow).To(BeNumerically(">=", 0))
+			Expect(dow).To(BeNumerically("<=", 6))
+		})
+
+		It("preserves TZ= prefix through resolution", func() {
+			sched, err := ParseCrontab("TZ=America/New_York 0~30 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			nyc, _ := time.LoadLocation("America/New_York")
+			Expect(spec.Location).To(Equal(nyc))
+		})
+
+		It("preserves non-random fields", func() {
+			sched, err := ParseCrontab("0~30 10 * * *")
+			Expect(err).ToNot(HaveOccurred())
+			spec := sched.(*cron.SpecSchedule)
+			Expect(spec.Hour & (1 << 10)).ToNot(BeZero())
+		})
+
+		It("resolves to a stable value across repeated Next calls", func() {
+			sched, err := ParseCrontab("0~30 * * * *")
+			Expect(err).ToNot(HaveOccurred())
+
+			ref := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)
+			first := sched.Next(ref)
+			for range 50 {
+				Expect(sched.Next(ref)).To(Equal(first))
+			}
+		})
+	})
+
+	Describe("error cases", func() {
+		It("rejects min > max", func() {
+			_, err := ParseCrontab("30~0 * * * *")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("beyond end"))
+		})
+
+		It("rejects value above field maximum", func() {
+			_, err := ParseCrontab("0~60 * * * *")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("above maximum"))
+		})
+
+		It("rejects value below field minimum", func() {
+			_, err := ParseCrontab("0 0 0~15 * *")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("below minimum"))
+		})
+
+		It("rejects ~ mixed with comma (list)", func() {
+			_, err := ParseCrontab("0~30,45 * * * *")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("cannot be combined"))
+		})
+
+		It("rejects ~ mixed with slash (step)", func() {
+			_, err := ParseCrontab("0~30/5 * * * *")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("cannot be combined"))
+		})
+
+		It("rejects @ descriptor with ~", func() {
+			_, err := ParseCrontab("@every 0~30m")
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("descriptor"))
+		})
+
+		It("rejects wrong number of fields", func() {
+			_, err := ParseCrontab("0~30 * *")
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("rejects non-numeric range values", func() {
+			_, err := ParseCrontab("a~b * * * *")
+			Expect(err).To(HaveOccurred())
+		})
+	})
+})
+
+// findSetBit returns the lowest bit position set in v, ignoring the starBit (bit 63).
+func findSetBit(v uint64) int {
+	v &^= 1 << 63 // clear starBit
+	for i := 0; i < 63; i++ {
+		if v&(1<<uint(i)) != 0 {
+			return i
+		}
+	}
+	return -1
+}
@@ -33,10 +33,11 @@ func (s *scheduler) Run(ctx context.Context) {
 }

 func (s *scheduler) Add(crontab string, cmd func()) (int, error) {
-	entryID, err := s.c.AddFunc(crontab, cmd)
+	schedule, err := ParseCrontab(crontab)
 	if err != nil {
 		return 0, err
 	}
+	entryID := s.c.Schedule(schedule, cron.FuncJob(cmd))
 	return int(entryID), nil
 }

@@ -1,19 +1,16 @@
 package scheduler

 import (
-	"sync"
 	"testing"
 	"time"

 	"github.com/navidrome/navidrome/log"
-	"github.com/navidrome/navidrome/tests"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/robfig/cron/v3"
 )

 func TestScheduler(t *testing.T) {
-	tests.Init(t, false)
 	log.SetLevel(log.LevelFatal)
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Scheduler Suite")
@@ -33,54 +30,49 @@ var _ = Describe("Scheduler", func() {
 	})

 	It("adds and executes a job", func() {
-		wg := sync.WaitGroup{}
-		wg.Add(1)
+		done := make(chan struct{})

-		executed := false
-		id, err := s.Add("@every 100ms", func() {
-			executed = true
-			wg.Done()
+		id, err := s.Add("@every 50ms", func() {
+			close(done)
 		})

 		Expect(err).ToNot(HaveOccurred())
 		Expect(id).ToNot(BeZero())

-		wg.Wait()
-		Expect(executed).To(BeTrue())
+		Eventually(done).Should(BeClosed())
+	})
+
+	It("adds a job with random ~ syntax", func() {
+		id, err := s.Add("0~59 * * * *", func() {})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(id).ToNot(BeZero())
+		s.Remove(id)
 	})

 	It("removes a job", func() {
-		// Use a WaitGroup to ensure the job executes once
-		wg := sync.WaitGroup{}
-		wg.Add(1)
+		done := make(chan struct{})

 		counter := 0
-		id, err := s.Add("@every 100ms", func() {
+		id, err := s.Add("@every 50ms", func() {
 			counter++
 			if counter == 1 {
-				wg.Done() // Signal that the job has executed once
+				close(done)
 			}
 		})
-
 		Expect(err).ToNot(HaveOccurred())
 		Expect(id).ToNot(BeZero())

-		// Wait for the job to execute at least once
-		wg.Wait()
-
 		// Verify job executed
+		Eventually(done).Should(BeClosed())
 		Expect(counter).To(Equal(1))

 		// Remove the job
 		s.Remove(id)

-		// Store the counter value
-		currentCount := counter
-
 		// Wait some time to ensure job doesn't execute again
 		time.Sleep(200 * time.Millisecond)

 		// Verify counter didn't increase
-		Expect(counter).To(Equal(currentCount))
+		Expect(counter).To(Equal(1))
 	})
 })