feat(cron): support timezone-aware cron scheduling
This commit is contained in:
parent
8206085f8f
commit
4746e1c760
3 changed files with 150 additions and 1 deletions
|
|
@ -8,6 +8,7 @@ import (
|
||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -264,7 +265,12 @@ func (cs *CronService) computeNextRun(schedule *CronSchedule, nowMS int64) *int6
|
||||||
}
|
}
|
||||||
|
|
||||||
// Use gronx to calculate next run time
|
// Use gronx to calculate next run time
|
||||||
now := time.UnixMilli(nowMS)
|
loc, err := resolveScheduleLocation(schedule.TZ)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("[cron] failed to load timezone %q: %v", schedule.TZ, err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
now := time.UnixMilli(nowMS).In(loc)
|
||||||
nextTime, err := gronx.NextTickAfter(schedule.Expr, now, false)
|
nextTime, err := gronx.NextTickAfter(schedule.Expr, now, false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("[cron] failed to compute next run for expr '%s': %v", schedule.Expr, err)
|
log.Printf("[cron] failed to compute next run for expr '%s': %v", schedule.Expr, err)
|
||||||
|
|
@ -278,6 +284,14 @@ func (cs *CronService) computeNextRun(schedule *CronSchedule, nowMS int64) *int6
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func resolveScheduleLocation(tz string) (*time.Location, error) {
|
||||||
|
trimmed := strings.TrimSpace(tz)
|
||||||
|
if trimmed == "" || strings.EqualFold(trimmed, "local") {
|
||||||
|
return time.Local, nil
|
||||||
|
}
|
||||||
|
return time.LoadLocation(trimmed)
|
||||||
|
}
|
||||||
|
|
||||||
func (cs *CronService) recomputeNextRuns() {
|
func (cs *CronService) recomputeNextRuns() {
|
||||||
now := time.Now().UnixMilli()
|
now := time.Now().UnixMilli()
|
||||||
for i := range cs.store.Jobs {
|
for i := range cs.store.Jobs {
|
||||||
|
|
@ -354,6 +368,9 @@ func (cs *CronService) AddJob(
|
||||||
defer cs.mu.Unlock()
|
defer cs.mu.Unlock()
|
||||||
|
|
||||||
now := time.Now().UnixMilli()
|
now := time.Now().UnixMilli()
|
||||||
|
if err := cs.validateSchedule(&schedule, now); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
// One-time tasks (at) should be deleted after execution
|
// One-time tasks (at) should be deleted after execution
|
||||||
deleteAfterRun := (schedule.Kind == "at")
|
deleteAfterRun := (schedule.Kind == "at")
|
||||||
|
|
@ -386,6 +403,42 @@ func (cs *CronService) AddJob(
|
||||||
return &job, nil
|
return &job, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (cs *CronService) validateSchedule(schedule *CronSchedule, nowMS int64) error {
|
||||||
|
if schedule == nil {
|
||||||
|
return fmt.Errorf("schedule is required")
|
||||||
|
}
|
||||||
|
|
||||||
|
switch schedule.Kind {
|
||||||
|
case "at":
|
||||||
|
if schedule.AtMS == nil {
|
||||||
|
return fmt.Errorf("at schedule requires atMs")
|
||||||
|
}
|
||||||
|
if *schedule.AtMS <= nowMS {
|
||||||
|
return fmt.Errorf("at schedule time must be in the future")
|
||||||
|
}
|
||||||
|
case "every":
|
||||||
|
if schedule.EveryMS == nil || *schedule.EveryMS <= 0 {
|
||||||
|
return fmt.Errorf("every schedule requires everyMs > 0")
|
||||||
|
}
|
||||||
|
case "cron":
|
||||||
|
expr := strings.TrimSpace(schedule.Expr)
|
||||||
|
if expr == "" {
|
||||||
|
return fmt.Errorf("cron schedule requires expr")
|
||||||
|
}
|
||||||
|
schedule.Expr = expr
|
||||||
|
if !cs.gronx.IsValid(expr) {
|
||||||
|
return fmt.Errorf("invalid cron expression: %s", expr)
|
||||||
|
}
|
||||||
|
if _, err := resolveScheduleLocation(schedule.TZ); err != nil {
|
||||||
|
return fmt.Errorf("invalid timezone %q: %w", schedule.TZ, err)
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unsupported schedule kind: %s", schedule.Kind)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (cs *CronService) UpdateJob(job *CronJob) error {
|
func (cs *CronService) UpdateJob(job *CronJob) error {
|
||||||
cs.mu.Lock()
|
cs.mu.Lock()
|
||||||
defer cs.mu.Unlock()
|
defer cs.mu.Unlock()
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,9 @@ import (
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/adhocore/gronx"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestSaveStore_FilePermissions(t *testing.T) {
|
func TestSaveStore_FilePermissions(t *testing.T) {
|
||||||
|
|
@ -36,3 +39,90 @@ func TestSaveStore_FilePermissions(t *testing.T) {
|
||||||
func int64Ptr(v int64) *int64 {
|
func int64Ptr(v int64) *int64 {
|
||||||
return &v
|
return &v
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestComputeNextRun_CronUsesScheduleTimezone(t *testing.T) {
|
||||||
|
cs := NewCronService(filepath.Join(t.TempDir(), "jobs.json"), nil)
|
||||||
|
|
||||||
|
// Use a fixed UTC reference to make this deterministic across environments.
|
||||||
|
nowMS := time.Date(2026, 2, 24, 0, 30, 0, 0, time.UTC).UnixMilli()
|
||||||
|
expr := "0 9 * * *"
|
||||||
|
|
||||||
|
utcSchedule := CronSchedule{
|
||||||
|
Kind: "cron",
|
||||||
|
Expr: expr,
|
||||||
|
TZ: "UTC",
|
||||||
|
}
|
||||||
|
shSchedule := CronSchedule{
|
||||||
|
Kind: "cron",
|
||||||
|
Expr: expr,
|
||||||
|
TZ: "Asia/Shanghai",
|
||||||
|
}
|
||||||
|
|
||||||
|
utcNextMS := cs.computeNextRun(&utcSchedule, nowMS)
|
||||||
|
if utcNextMS == nil {
|
||||||
|
t.Fatalf("expected UTC next run, got nil")
|
||||||
|
}
|
||||||
|
shNextMS := cs.computeNextRun(&shSchedule, nowMS)
|
||||||
|
if shNextMS == nil {
|
||||||
|
t.Fatalf("expected Asia/Shanghai next run, got nil")
|
||||||
|
}
|
||||||
|
|
||||||
|
expectedUTC, err := gronx.NextTickAfter(expr, time.UnixMilli(nowMS).In(time.UTC), false)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to compute expected UTC tick: %v", err)
|
||||||
|
}
|
||||||
|
shLoc, err := time.LoadLocation("Asia/Shanghai")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to load Asia/Shanghai location: %v", err)
|
||||||
|
}
|
||||||
|
expectedSH, err := gronx.NextTickAfter(expr, time.UnixMilli(nowMS).In(shLoc), false)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to compute expected Asia/Shanghai tick: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if got, want := *utcNextMS, expectedUTC.UnixMilli(); got != want {
|
||||||
|
t.Fatalf("UTC next run mismatch: got %d, want %d", got, want)
|
||||||
|
}
|
||||||
|
if got, want := *shNextMS, expectedSH.UnixMilli(); got != want {
|
||||||
|
t.Fatalf("Asia/Shanghai next run mismatch: got %d, want %d", got, want)
|
||||||
|
}
|
||||||
|
if *utcNextMS == *shNextMS {
|
||||||
|
t.Fatalf("expected timezone-specific next run to differ, both were %d", *utcNextMS)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestComputeNextRun_CronInvalidTimezone(t *testing.T) {
|
||||||
|
cs := NewCronService(filepath.Join(t.TempDir(), "jobs.json"), nil)
|
||||||
|
|
||||||
|
nowMS := time.Date(2026, 2, 24, 0, 30, 0, 0, time.UTC).UnixMilli()
|
||||||
|
schedule := CronSchedule{
|
||||||
|
Kind: "cron",
|
||||||
|
Expr: "*/5 * * * *",
|
||||||
|
TZ: "Mars/OlympusMons",
|
||||||
|
}
|
||||||
|
|
||||||
|
next := cs.computeNextRun(&schedule, nowMS)
|
||||||
|
if next != nil {
|
||||||
|
t.Fatalf("expected nil for invalid timezone, got %d", *next)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAddJob_CronInvalidTimezoneReturnsError(t *testing.T) {
|
||||||
|
cs := NewCronService(filepath.Join(t.TempDir(), "jobs.json"), nil)
|
||||||
|
|
||||||
|
_, err := cs.AddJob(
|
||||||
|
"bad-tz",
|
||||||
|
CronSchedule{
|
||||||
|
Kind: "cron",
|
||||||
|
Expr: "*/5 * * * *",
|
||||||
|
TZ: "Mars/OlympusMons",
|
||||||
|
},
|
||||||
|
"hello",
|
||||||
|
false,
|
||||||
|
"cli",
|
||||||
|
"direct",
|
||||||
|
)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("expected error for invalid timezone")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -84,6 +84,10 @@ func (t *CronTool) Parameters() map[string]any {
|
||||||
"type": "string",
|
"type": "string",
|
||||||
"description": "Cron expression for complex recurring schedules (e.g., '0 9 * * *' for daily at 9am). Use this for complex recurring schedules.",
|
"description": "Cron expression for complex recurring schedules (e.g., '0 9 * * *' for daily at 9am). Use this for complex recurring schedules.",
|
||||||
},
|
},
|
||||||
|
"timezone": map[string]any{
|
||||||
|
"type": "string",
|
||||||
|
"description": "Optional IANA timezone for cron_expr (e.g., 'Asia/Shanghai'). Defaults to local timezone.",
|
||||||
|
},
|
||||||
"job_id": map[string]any{
|
"job_id": map[string]any{
|
||||||
"type": "string",
|
"type": "string",
|
||||||
"description": "Job ID (for remove/enable/disable)",
|
"description": "Job ID (for remove/enable/disable)",
|
||||||
|
|
@ -149,6 +153,7 @@ func (t *CronTool) addJob(args map[string]any) *ToolResult {
|
||||||
atSeconds, hasAt := args["at_seconds"].(float64)
|
atSeconds, hasAt := args["at_seconds"].(float64)
|
||||||
everySeconds, hasEvery := args["every_seconds"].(float64)
|
everySeconds, hasEvery := args["every_seconds"].(float64)
|
||||||
cronExpr, hasCron := args["cron_expr"].(string)
|
cronExpr, hasCron := args["cron_expr"].(string)
|
||||||
|
timezone, _ := args["timezone"].(string)
|
||||||
|
|
||||||
// Priority: at_seconds > every_seconds > cron_expr
|
// Priority: at_seconds > every_seconds > cron_expr
|
||||||
if hasAt {
|
if hasAt {
|
||||||
|
|
@ -167,6 +172,7 @@ func (t *CronTool) addJob(args map[string]any) *ToolResult {
|
||||||
schedule = cron.CronSchedule{
|
schedule = cron.CronSchedule{
|
||||||
Kind: "cron",
|
Kind: "cron",
|
||||||
Expr: cronExpr,
|
Expr: cronExpr,
|
||||||
|
TZ: timezone,
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
return ErrorResult("one of at_seconds, every_seconds, or cron_expr is required")
|
return ErrorResult("one of at_seconds, every_seconds, or cron_expr is required")
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue