Documentation
¶
Index ¶
- Variables
- func NextCronTime(expr string, after time.Time) *time.Time
- func ValidateCronExpr(expr string) error
- type Scheduler
- func (s *Scheduler) MarkDone(ctx context.Context, scheduleID, jobID string, status model.JobStatus)
- func (s *Scheduler) RecordTrigger(ctx context.Context, scheduleID, jobID, cronExpr string, triggerErr error)
- func (s *Scheduler) Reload()
- func (s *Scheduler) RunNow(ctx context.Context, task *model.ScheduledTask) (string, error)
- func (s *Scheduler) Start(ctx context.Context)
- func (s *Scheduler) Stop()
- type Service
- type TriggerFunc
Constants ¶
This section is empty.
Variables ¶
var ( ErrGraphWorkflowRequired = errors.New("graph workflow id is required") ErrGraphWorkflowNotFound = errors.New("graph workflow not found") )
var ErrScheduleMaxConcurrent = errors.New("schedule max concurrent reached")
ErrScheduleMaxConcurrent is returned by RunNow when a manual trigger is refused because the schedule already has maxConcurrent runs in flight. The HTTP layer maps it to 409 Conflict and, unlike a real trigger failure, it is NOT recorded on the task (no slot was taken, no run was attempted).
Functions ¶
func NextCronTime ¶
NextCronTime calculates the next fire time after `after` for the given cron expression.
func ValidateCronExpr ¶
ValidateCronExpr checks if a cron expression is syntactically valid.
Types ¶
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler manages cron-based execution of scheduled tasks.
func NewScheduler ¶
func NewScheduler(svc Service, trigger TriggerFunc) *Scheduler
func (*Scheduler) MarkDone ¶
MarkDone updates a task's LastStatus when its job completes and decrements the running count so the scheduler can trigger new runs.
jobID is the ID of the job that actually finished. It is NOT read from task.LastRunJobID: with maxConcurrent>1 (or a fast-completing job whose RecordTrigger hasn't landed yet) LastRunJobID can point at a sibling run, and using it here would make logs and status reflect the wrong job.
func (*Scheduler) RecordTrigger ¶
func (s *Scheduler) RecordTrigger(ctx context.Context, scheduleID, jobID, cronExpr string, triggerErr error)
RecordTrigger atomically updates a scheduled task's status after a trigger attempt. Called by both tryTrigger (cron) and RunNow (manual) paths to avoid duplicated logic.
func (*Scheduler) Reload ¶
func (s *Scheduler) Reload()
Reload is a no-op signal for future use. The tick-based scheduler re-reads all tasks every minute, so changes take effect automatically.
func (*Scheduler) RunNow ¶
RunNow manually triggers a task, respecting maxConcurrent. Returns the job ID or an error.
RunNow records the trigger outcome itself (success or failure) via RecordTrigger, so cron (tryTrigger) and manual paths share one failure-record rule per the design's "手动立即运行与 cron 触发统一失败记录". An over-limit refusal returns ErrScheduleMaxConcurrent and is intentionally NOT recorded: no slot was taken and no run was attempted. RecordTrigger runs on a background context so a disconnected HTTP request can't cancel the status write-back.
type Service ¶
type Service interface {
Create(ctx context.Context, req *model.CreateScheduleRequest) (*model.ScheduledTask, error)
Get(ctx context.Context, id string) (*model.ScheduledTask, error)
List(ctx context.Context) ([]*model.ScheduledTask, error)
ListByWorkspace(ctx context.Context, wsID string) ([]*model.ScheduledTask, error)
Update(ctx context.Context, id string, req *model.UpdateScheduleRequest) (*model.ScheduledTask, error)
Toggle(ctx context.Context, id string) (*model.ScheduledTask, error)
Delete(ctx context.Context, id string) error
SaveState(ctx context.Context, task *model.ScheduledTask) error
}
Service provides CRUD operations for scheduled tasks.
func NewService ¶
type TriggerFunc ¶
TriggerFunc is called by the scheduler to create and start a job for the given task.
Source Files