feat: implement distributed scheduler using Redis sorted sets #71
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| package queue | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strconv" | ||
| "strings" | ||
| "time" | ||
| ) | ||
|
|
||
| // parsePattern parses a cron-like pattern and returns the polling interval. | ||
| // Supports two formats: | ||
| // 1. @every <duration> format (e.g., "@every 1s", "@every 5m") | ||
| // 2. Cron expressions (e.g., "*/5 * * * *", "0 */2 * * *") | ||
| // Returns an error if the pattern is invalid or unsupported. | ||
| func parsePattern(pattern string) (time.Duration, error) { | ||
| if pattern == "" { | ||
| return 0, fmt.Errorf("pattern cannot be empty") | ||
| } | ||
|
|
||
| // Trim whitespace | ||
| pattern = strings.TrimSpace(pattern) | ||
|
|
||
| // Check for @every prefix | ||
| if strings.HasPrefix(pattern, "@every ") { | ||
| return parseEveryPattern(pattern) | ||
| } | ||
|
|
||
| // Try to parse as cron expression | ||
| return parseCronPattern(pattern) | ||
| } | ||
|
|
||
| // parseEveryPattern parses @every <duration> format patterns. | ||
| func parseEveryPattern(pattern string) (time.Duration, error) { | ||
| // Extract the duration part after "@every " | ||
| durationStr := strings.TrimSpace(strings.TrimPrefix(pattern, "@every ")) | ||
| if durationStr == "" { | ||
| return 0, fmt.Errorf("missing duration in pattern: %s", pattern) | ||
| } | ||
|
|
||
| // Parse the duration using time.ParseDuration | ||
| duration, err := time.ParseDuration(durationStr) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("invalid duration '%s': %w", durationStr, err) | ||
| } | ||
|
|
||
| // Validate that duration is positive | ||
| if duration <= 0 { | ||
| return 0, fmt.Errorf("duration must be positive, got: %s", duration) | ||
| } | ||
|
|
||
| return duration, nil | ||
| } | ||
|
|
||
| // parseCronPattern parses cron expressions and calculates the polling interval. | ||
| // Supports standard 5-field cron format: minute hour day month weekday | ||
| // Examples: | ||
| // - "*/5 * * * *" → every 5 minutes | ||
| // - "0 * * * *" → every hour | ||
| // - "0 0 * * *" → every day (24 hours) | ||
| // - "0 0 * * 0" → every week (7 days) | ||
| func parseCronPattern(pattern string) (time.Duration, error) { | ||
| fields := strings.Fields(pattern) | ||
| if len(fields) != 5 { | ||
| return 0, fmt.Errorf("invalid cron expression: expected 5 fields, got %d in '%s'", len(fields), pattern) | ||
| } | ||
|
|
||
| minute, hour, day, month, weekday := fields[0], fields[1], fields[2], fields[3], fields[4] | ||
|
|
||
| // Parse minute field for */N patterns | ||
| if strings.HasPrefix(minute, "*/") { | ||
| intervalStr := strings.TrimPrefix(minute, "*/") | ||
| interval, err := strconv.Atoi(intervalStr) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("invalid minute interval '%s': %w", intervalStr, err) | ||
| } | ||
| if interval <= 0 || interval > 59 { | ||
| return 0, fmt.Errorf("minute interval must be between 1 and 59, got %d", interval) | ||
| } | ||
| return time.Duration(interval) * time.Minute, nil | ||
| } | ||
|
|
||
| // Parse hour field for */N patterns | ||
| if strings.HasPrefix(hour, "*/") { | ||
| intervalStr := strings.TrimPrefix(hour, "*/") | ||
| interval, err := strconv.Atoi(intervalStr) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("invalid hour interval '%s': %w", intervalStr, err) | ||
| } | ||
| if interval <= 0 || interval > 23 { | ||
| return 0, fmt.Errorf("hour interval must be between 1 and 23, got %d", interval) | ||
| } | ||
| return time.Duration(interval) * time.Hour, nil | ||
| } | ||
|
|
||
| // Hourly: "0 * * * *" or "N * * * *" | ||
| if hour == "*" && day == "*" && month == "*" && weekday == "*" { | ||
| return 1 * time.Hour, nil | ||
| } | ||
|
|
||
| // Daily: "0 0 * * *" or "N N * * *" | ||
| if day == "*" && month == "*" && weekday == "*" { | ||
| return 24 * time.Hour, nil | ||
| } | ||
|
|
||
| // Weekly: "0 0 * * N" (specific weekday) | ||
| if day == "*" && month == "*" && weekday != "*" { | ||
| return 7 * 24 * time.Hour, nil | ||
| } | ||
|
|
||
| // Monthly: "0 0 N * *" (specific day of month) | ||
| if month == "*" && weekday == "*" && day != "*" { | ||
| return 30 * 24 * time.Hour, nil // Approximate as 30 days | ||
| } | ||
|
|
||
| return 0, fmt.Errorf("unsupported cron pattern: %s (consider using @every <duration> format)", pattern) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| package queue_test | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/redis/go-redis/v9" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/tinh-tinh/queue/v2" | ||
| ) | ||
|
|
||
| func Test_ParsePattern_Valid(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| pattern string | ||
| expected time.Duration | ||
| }{ | ||
| // @every patterns | ||
| { | ||
| name: "1 second", | ||
| pattern: "@every 1s", | ||
| expected: 1 * time.Second, | ||
| }, | ||
| { | ||
| name: "5 seconds", | ||
| pattern: "@every 5s", | ||
| expected: 5 * time.Second, | ||
| }, | ||
| { | ||
| name: "1 minute", | ||
| pattern: "@every 1m", | ||
| expected: 1 * time.Minute, | ||
| }, | ||
| { | ||
| name: "5 minutes", | ||
| pattern: "@every 5m", | ||
| expected: 5 * time.Minute, | ||
| }, | ||
| { | ||
| name: "1 hour", | ||
| pattern: "@every 1h", | ||
| expected: 1 * time.Hour, | ||
| }, | ||
| { | ||
| name: "complex duration", | ||
| pattern: "@every 1h30m45s", | ||
| expected: 1*time.Hour + 30*time.Minute + 45*time.Second, | ||
| }, | ||
| { | ||
| name: "with extra spaces", | ||
| pattern: "@every 5s ", | ||
| expected: 5 * time.Second, | ||
| }, | ||
| { | ||
| name: "milliseconds", | ||
| pattern: "@every 500ms", | ||
| expected: 500 * time.Millisecond, | ||
| }, | ||
| // Cron patterns | ||
| { | ||
| name: "every 5 minutes (cron)", | ||
| pattern: "*/5 * * * *", | ||
| expected: 5 * time.Minute, | ||
| }, | ||
| { | ||
| name: "every 15 minutes (cron)", | ||
| pattern: "*/15 * * * *", | ||
| expected: 15 * time.Minute, | ||
| }, | ||
| { | ||
| name: "every 2 hours (cron)", | ||
| pattern: "0 */2 * * *", | ||
| expected: 2 * time.Hour, | ||
| }, | ||
| { | ||
| name: "every 6 hours (cron)", | ||
| pattern: "0 */6 * * *", | ||
| expected: 6 * time.Hour, | ||
| }, | ||
| { | ||
| name: "hourly (cron)", | ||
| pattern: "0 * * * *", | ||
| expected: 1 * time.Hour, | ||
| }, | ||
| { | ||
| name: "daily (cron)", | ||
| pattern: "0 0 * * *", | ||
| expected: 24 * time.Hour, | ||
| }, | ||
| { | ||
| name: "weekly (cron)", | ||
| pattern: "0 0 * * 0", | ||
| expected: 7 * 24 * time.Hour, | ||
| }, | ||
| { | ||
| name: "monthly (cron)", | ||
| pattern: "0 0 1 * *", | ||
| expected: 30 * 24 * time.Hour, | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| // Test indirectly through Queue creation | ||
| q := queue.New("test_pattern_"+tt.name, &queue.Options{ | ||
| Connect: &redis.Options{ | ||
| Addr: "localhost:6379", | ||
| Password: "", | ||
| DB: 0, | ||
| }, | ||
| Workers: 1, | ||
| RetryFailures: 0, | ||
| Pattern: tt.pattern, | ||
| // Don't set ScheduleInterval to force pattern parsing | ||
| }) | ||
| require.NotNil(t, q) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| func Test_ParsePattern_Invalid(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| pattern string | ||
| }{ | ||
| { | ||
| name: "empty pattern", | ||
| pattern: "", | ||
| }, | ||
|
Comment on lines
+126
to
+129
There was a problem hiding this comment. Empty pattern test doesn't exercise the fallback behavior. Based on Consider removing this case or documenting that empty pattern means "no scheduler" rather than "fallback to default." 🤖 Prompt for AI Agents |
||
| { | ||
| name: "missing duration", | ||
| pattern: "@every ", | ||
| }, | ||
| { | ||
| name: "invalid duration", | ||
| pattern: "@every abc", | ||
| }, | ||
| { | ||
| name: "negative duration", | ||
| pattern: "@every -5s", | ||
| }, | ||
| { | ||
| name: "invalid cron - too few fields", | ||
| pattern: "*/5 * *", | ||
| }, | ||
| { | ||
| name: "invalid cron - too many fields", | ||
| pattern: "*/5 * * * * *", | ||
| }, | ||
| { | ||
| name: "invalid cron - bad minute interval", | ||
| pattern: "*/abc * * * *", | ||
| }, | ||
| { | ||
| name: "invalid cron - minute out of range", | ||
| pattern: "*/60 * * * *", | ||
| }, | ||
| { | ||
| name: "invalid cron - hour out of range", | ||
| pattern: "0 */24 * * *", | ||
| }, | ||
| { | ||
| name: "unsupported cron pattern", | ||
| pattern: "5,10,15 * * * *", | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| // These should fall back to default 5s interval with a warning log | ||
| q := queue.New("test_invalid_"+tt.name, &queue.Options{ | ||
| Connect: &redis.Options{ | ||
| Addr: "localhost:6379", | ||
| Password: "", | ||
| DB: 0, | ||
| }, | ||
| Workers: 1, | ||
| RetryFailures: 0, | ||
| Pattern: tt.pattern, | ||
| DisableLog: true, // Disable logs to avoid clutter in tests | ||
| }) | ||
| require.NotNil(t, q) | ||
| }) | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🧩 Analysis chain
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 975
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 2113
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 1108
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 1551
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 4272
🏁 Script executed:
Repository: tinh-tinh/queue
Length of output: 2115
ActiveStatusshould not be included inIsReady().A job in
ActiveStatusrepresents one currently being or previously being processed. Including it in the ready check violates job state semantics and could cause duplicate processing.The job lifecycle should be:
WaitStatus→ActiveStatus(duringProcess()) →CompletedStatus/FailedStatus/DelayedStatus(final states). OnlyWaitStatusjobs should be returned fromIsReady(). Delayed jobs are handled separately in theRetry()method and should not be included here.The comment "Jobs are ready if they are waiting or active" is misleading—only waiting jobs are ready for initial processing.
Suggested fix
📝 Committable suggestion
🤖 Prompt for AI Agents