Skip to content

Commit cdfbf1b

Browse files
hyp3rdhyp3rd
authored
Merge pull request #87 from hyp3rd/feat/admin-ui
feat(cron): migrate from robfig/cron/v3 to hyp3rd/cron/v4
2 parents 806b27a + bceb7a1 commit cdfbf1b

5 files changed

Lines changed: 45 additions & 31 deletions

File tree

‎admin_taskmanager.go‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@ import (
88
"time"
99

1010
"github.com/google/uuid"
11+
"github.com/hyp3rd/cron/v4"
1112
"github.com/hyp3rd/ewrap"
12-
"github.com/robfig/cron/v3"
1313
)
1414

1515
func (tm *TaskManager) adminBackend() (AdminBackend, error) {
@@ -124,7 +124,11 @@ func (tm *TaskManager) AdminSchedules(ctx context.Context) ([]AdminSchedule, err
124124

125125
results := make([]AdminSchedule, 0, len(tm.cronEntries))
126126
for _, entry := range tm.cron.Entries() {
127-
name := nameByID[entry.ID]
127+
name := entry.Name
128+
if name == "" {
129+
name = nameByID[entry.ID]
130+
}
131+
128132
if name == "" {
129133
continue
130134
}
@@ -474,7 +478,7 @@ func (tm *TaskManager) AdminCreateSchedule(ctx context.Context, spec AdminSchedu
474478
tm.cron.Remove(existing)
475479
}
476480

477-
entryID := tm.cron.Schedule(schedule, cron.FuncJob(tm.cronJob(name)))
481+
entryID := tm.scheduleCronEntry(name, schedule)
478482
tm.cronEntries[name] = entryID
479483
tm.cronSpecs[name] = cronSpec{Spec: specValue, Durable: factory.Durable}
480484

‎cron.go‎

Lines changed: 34 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,14 @@ package worker
22

33
import (
44
"context"
5+
"errors"
56
"log/slog"
67
"strings"
78
"time"
89

910
"github.com/google/uuid"
11+
"github.com/hyp3rd/cron/v4"
1012
"github.com/hyp3rd/ewrap"
11-
"github.com/robfig/cron/v3"
1213
)
1314

1415
const errParseCronSchedule = "parse cron schedule"
@@ -66,7 +67,7 @@ func (tm *TaskManager) RegisterCronTask(
6667
Origin: cronFactoryOriginUser,
6768
}
6869

69-
entryID := tm.cron.Schedule(schedule, cron.FuncJob(tm.cronJob(normalized)))
70+
entryID := tm.scheduleCronEntry(normalized, schedule)
7071

7172
tm.cronEntries[normalized] = entryID
7273
tm.cronSpecs[normalized] = cronSpec{Spec: strings.TrimSpace(spec), Durable: false}
@@ -103,32 +104,34 @@ func (tm *TaskManager) RegisterDurableCronTask(
103104
Origin: cronFactoryOriginUser,
104105
}
105106

106-
entryID := tm.cron.Schedule(schedule, cron.FuncJob(tm.cronJob(normalized)))
107+
entryID := tm.scheduleCronEntry(normalized, schedule)
107108

108109
tm.cronEntries[normalized] = entryID
109110
tm.cronSpecs[normalized] = cronSpec{Spec: strings.TrimSpace(spec), Durable: true}
110111

111112
return nil
112113
}
113114

114-
func (tm *TaskManager) cronJob(name string) func() {
115-
return func() {
116-
if tm.skipCronTick() {
117-
return
115+
func (tm *TaskManager) cronJob(name string) func(context.Context) error {
116+
return func(ctx context.Context) error {
117+
if tm.skipCronTick(ctx) {
118+
return nil
118119
}
119120

120121
spec, factory, ok := tm.cronSpecAndFactory(name)
121122
if !ok {
122-
return
123+
return nil
123124
}
124125

125126
if factory.Durable {
126-
tm.runDurableCron(name, spec, factory)
127+
tm.runDurableCron(ctx, name, spec, factory)
127128

128-
return
129+
return nil
129130
}
130131

131-
tm.runInMemoryCron(name, spec, factory)
132+
tm.runInMemoryCron(ctx, name, spec, factory)
133+
134+
return nil
132135
}
133136
}
134137

@@ -145,8 +148,8 @@ func (tm *TaskManager) cronSpecAndFactory(name string) (cronSpec, cronFactory, b
145148
return spec, factory, true
146149
}
147150

148-
func (tm *TaskManager) runDurableCron(name string, spec cronSpec, factory cronFactory) {
149-
task, err := factory.DurableFactory(tm.ctx)
151+
func (tm *TaskManager) runDurableCron(ctx context.Context, name string, spec cronSpec, factory cronFactory) {
152+
task, err := factory.DurableFactory(ctx)
150153
if err != nil {
151154
cronLogError("cron durable task factory", name, err)
152155

@@ -187,15 +190,15 @@ func (tm *TaskManager) runDurableCron(name string, spec cronSpec, factory cronFa
187190

188191
tm.noteCronRun(runInfo)
189192

190-
err = tm.RegisterDurableTask(tm.ctx, task)
193+
err = tm.RegisterDurableTask(ctx, task)
191194
if err != nil {
192195
tm.dropCronRun(task.ID)
193196
cronLogError("cron register durable task", name, err)
194197
}
195198
}
196199

197-
func (tm *TaskManager) runInMemoryCron(name string, spec cronSpec, factory cronFactory) {
198-
task, err := factory.TaskFactory(tm.ctx)
200+
func (tm *TaskManager) runInMemoryCron(ctx context.Context, name string, spec cronSpec, factory cronFactory) {
201+
task, err := factory.TaskFactory(ctx)
199202
if err != nil {
200203
cronLogError("cron task factory", name, err)
201204

@@ -215,7 +218,7 @@ func (tm *TaskManager) runInMemoryCron(name string, spec cronSpec, factory cronF
215218
runInfo := cronRunInfoFromTask(name, spec, task, tm.defaultQueue)
216219
tm.noteCronRun(runInfo)
217220

218-
err = tm.RegisterTask(tm.ctx, task)
221+
err = tm.RegisterTask(ctx, task)
219222
if err != nil {
220223
tm.dropCronRun(task.ID)
221224
cronLogError("cron register task", name, err)
@@ -258,8 +261,12 @@ func (tm *TaskManager) prepareCronRegistration(
258261
return name, schedule, nil
259262
}
260263

261-
func (tm *TaskManager) skipCronTick() bool {
262-
return tm.ctx.Err() != nil || !tm.accepting.Load()
264+
func (tm *TaskManager) scheduleCronEntry(name string, schedule cron.Schedule) cron.EntryID {
265+
return tm.cron.ScheduleNamed(name, schedule, cron.FuncJob(tm.cronJob(name)))
266+
}
267+
268+
func (tm *TaskManager) skipCronTick(ctx context.Context) bool {
269+
return ctx.Err() != nil || tm.ctx.Err() != nil || !tm.accepting.Load()
263270
}
264271

265272
// UnregisterCronTask removes a cron job by name.
@@ -295,7 +302,7 @@ func (tm *TaskManager) initCron() {
295302
tm.cronLoc = location
296303
}
297304

298-
parser := cron.NewParser(
305+
parser := cron.NewSpecParser(
299306
cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
300307
)
301308
tm.cron = cron.New(cron.WithLocation(location), cron.WithParser(parser))
@@ -306,7 +313,7 @@ func (tm *TaskManager) startCron() {
306313
defer tm.cronMu.Unlock()
307314

308315
if tm.cron != nil {
309-
tm.cron.Start()
316+
tm.cron.Start(tm.ctx)
310317
}
311318
}
312319

@@ -315,7 +322,10 @@ func (tm *TaskManager) stopCron() {
315322
defer tm.cronMu.Unlock()
316323

317324
if tm.cron != nil {
318-
tm.cron.Stop()
325+
err := tm.cron.Stop(tm.ctx)
326+
if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
327+
cronLogError("cron stop", "scheduler", err)
328+
}
319329
}
320330
}
321331

@@ -358,13 +368,13 @@ func parseCronSpec(spec string, location *time.Location) (cron.Schedule, error)
358368
}
359369

360370
func cronParserStandard(_ *time.Location) cron.Parser {
361-
return cron.NewParser(
371+
return cron.NewSpecParser(
362372
cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
363373
)
364374
}
365375

366376
func cronParserSeconds(_ *time.Location) cron.Parser {
367-
return cron.NewParser(
377+
return cron.NewSpecParser(
368378
cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
369379
)
370380
}

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@ go 1.26.2
55
require (
66
github.com/goccy/go-json v0.10.6
77
github.com/google/uuid v1.6.0
8+
github.com/hyp3rd/cron/v4 v4.0.0
89
github.com/hyp3rd/ewrap v1.3.9
910
github.com/hyp3rd/sectools v1.2.4
1011
github.com/redis/rueidis v1.0.73
11-
github.com/robfig/cron/v3 v3.0.1
1212
github.com/spf13/cobra v1.10.2
1313
go.opentelemetry.io/otel/metric v1.43.0
1414
go.opentelemetry.io/otel/sdk/metric v1.43.0

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
1616
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
1717
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
1818
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
19+
github.com/hyp3rd/cron/v4 v4.0.0 h1:S5cLICFWyxnsZCn/96KGd65I1poqg3lUn7hpRCx9ZLA=
20+
github.com/hyp3rd/cron/v4 v4.0.0/go.mod h1:4TQSY8fjNp05ajGxS2jFNdsHxBLAFk91snCRjW7z1v8=
1921
github.com/hyp3rd/ewrap v1.3.9 h1:4vtnxji/aJdnyR2dfl93R/uYcGrNdi93EbV/r5BYalk=
2022
github.com/hyp3rd/ewrap v1.3.9/go.mod h1:2AgfjKPZjfBxvlTrbdWrNZzxV3jqmcOHg38aKyXvxpQ=
2123
github.com/hyp3rd/hyperlogger v0.1.7 h1:v2ffVH/I/jqGTuPsvxoUjVk4Hn1vfhdDParw/ua+VQQ=
@@ -34,8 +36,6 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
3436
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
3537
github.com/redis/rueidis v1.0.73 h1:0Enrg0VuMdaYyNDDj0lLIheWY0uybCeQOh+jTp2GG3M=
3638
github.com/redis/rueidis v1.0.73/go.mod h1:lfdcZzJ1oKGKL37vh9fO3ymwt+0TdjkkUCJxbgpmcgQ=
37-
github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
38-
github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
3939
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
4040
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
4141
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=

‎worker.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,8 @@ import (
99
"time"
1010

1111
"github.com/google/uuid"
12+
"github.com/hyp3rd/cron/v4"
1213
"github.com/hyp3rd/ewrap"
13-
"github.com/robfig/cron/v3"
1414
"golang.org/x/time/rate"
1515
)
1616

0 commit comments

Comments
 (0)