From d0e1cc4ad6c4351068bc11c20f0e1c3d58e0f42a Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Tue, 4 Aug 2026 13:51:10 +0100 Subject: [PATCH] feat: cron arithmetic and persisted workflow schedules --- server/go.mod | 7 ++- server/go.sum | 2 + server/internal/models/workflow.go | 18 ++++++ server/internal/services/workflows.go | 48 ++++++++++++++++ server/internal/workflowsched/cron.go | 82 +++++++++++++++++++++++++++ 5 files changed, 155 insertions(+), 2 deletions(-) create mode 100644 server/internal/workflowsched/cron.go diff --git a/server/go.mod b/server/go.mod index a48fe08..d1ccff3 100644 --- a/server/go.mod +++ b/server/go.mod @@ -14,9 +14,13 @@ require ( google.golang.org/grpc v1.64.0 ) -require github.com/hyperboloide/lk v0.0.0-20251220053519-b291812e3216 // indirect +require ( + github.com/hyperboloide/lk v0.0.0-20251220053519-b291812e3216 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect +) require ( + gitea.hostxtra.co.uk/mrhid6/vantage/shared v0.0.0 github.com/bytedance/sonic v1.11.6 // indirect github.com/bytedance/sonic/loader v0.1.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect @@ -38,7 +42,6 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect - gitea.hostxtra.co.uk/mrhid6/vantage/shared v0.0.0 github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/sirupsen/logrus v1.4.2 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect diff --git a/server/go.sum b/server/go.sum index 27bbea6..b866001 100644 --- a/server/go.sum +++ b/server/go.sum @@ -70,6 +70,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w= github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4= github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/server/internal/models/workflow.go b/server/internal/models/workflow.go index 74e1906..333ec97 100644 --- a/server/internal/models/workflow.go +++ b/server/internal/models/workflow.go @@ -44,6 +44,20 @@ type StepOverride struct { SecretRefs []string `bson:"secret_refs,omitempty" json:"secret_refs,omitempty"` } +type Schedule struct { + Enabled bool `bson:"enabled" json:"enabled"` + Cron string `bson:"cron" json:"cron"` // 5-field: minute hour dom month dow + TZ string `bson:"tz" json:"tz"` // IANA name, e.g. Europe/London +} + +// Skip records why an occurrence did not run. Recording a reason nobody reads +// is the same as not recording one, so this is surfaced in the UI. +type Skip struct { + Reason string `bson:"reason" json:"reason"` // "missed" | "already_running" + Due time.Time `bson:"due" json:"due"` + At time.Time `bson:"at" json:"at"` +} + type Workflow struct { ID bson.ObjectID `bson:"_id,omitempty" json:"-"` InstanceID string `bson:"instance_id" json:"instance_id"` @@ -52,6 +66,10 @@ type Workflow struct { TargetServerIDs []string `bson:"target_server_ids" json:"target_server_ids"` TargetTags map[string]string `bson:"target_tags,omitempty" json:"target_tags,omitempty"` Steps []WorkflowStepRef `bson:"steps" json:"steps"` + Schedule *Schedule `bson:"schedule,omitempty" json:"schedule,omitempty"` + NextRunAt *time.Time `bson:"next_run_at,omitempty" json:"next_run_at,omitempty"` + LastRunAt *time.Time `bson:"last_run_at,omitempty" json:"last_run_at,omitempty"` + LastSkipped *Skip `bson:"last_skipped,omitempty" json:"last_skipped,omitempty"` CreatedAt time.Time `bson:"created_at" json:"created_at"` UpdatedAt time.Time `bson:"updated_at" json:"updated_at"` } diff --git a/server/internal/services/workflows.go b/server/internal/services/workflows.go index 6161483..af892fb 100644 --- a/server/internal/services/workflows.go +++ b/server/internal/services/workflows.go @@ -8,6 +8,7 @@ import ( "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched" "github.com/google/uuid" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" @@ -60,6 +61,11 @@ func EnsureWorkflowIndexes() error { }); err != nil { return err } + if _, err := db.Col("workflows").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "next_run_at", Value: 1}}, + }); err != nil { + return err + } if err := EnsureLogIndexes(); err != nil { return err } @@ -321,3 +327,45 @@ func DeleteWorkflow(instanceID, id string) error { _, err := db.Col("workflows").DeleteOne(ctx, bson.M{"workflow_id": id, "instance_id": instanceID}) return err } + +// SetSchedule validates and stores a workflow's schedule, computing the first +// occurrence. next_run_at is persisted rather than held in memory: a leader +// handover between computing an occurrence and firing it would otherwise lose +// it or fire it twice. +// +// Passing s == nil, or a disabled schedule, clears next_run_at so the +// scheduler's query stops matching the document at all. +func SetSchedule(instanceID, workflowID string, s *models.Schedule) (*time.Time, error) { + ctx, cancel := wfCtx() + defer cancel() + + set := bson.M{"schedule": s, "updated_at": time.Now()} + unset := bson.M{} + + var next *time.Time + if s != nil && s.Enabled { + at, err := workflowsched.NextOccurrence(s.Cron, s.TZ, time.Now()) + if err != nil { + return nil, err + } + next = &at + set["next_run_at"] = at + } else { + unset["next_run_at"] = "" + } + + update := bson.M{"$set": set} + if len(unset) > 0 { + update["$unset"] = unset + } + + res, err := db.Col("workflows").UpdateOne(ctx, + bson.M{"workflow_id": workflowID, "instance_id": instanceID}, update) + if err != nil { + return nil, err + } + if res.MatchedCount == 0 { + return nil, mongo.ErrNoDocuments + } + return next, nil +} diff --git a/server/internal/workflowsched/cron.go b/server/internal/workflowsched/cron.go new file mode 100644 index 0000000..17c1d59 --- /dev/null +++ b/server/internal/workflowsched/cron.go @@ -0,0 +1,82 @@ +// Package workflowsched fires workflow runs on a cron schedule. +// +// Only robfig/cron's parser is used — Parse and Next. Its own scheduler is +// not, because this work runs under the housekeeping leader lock and has to +// stop the moment leadership is lost. +package workflowsched + +import ( + "errors" + "fmt" + "time" + + "github.com/robfig/cron/v3" +) + +// ErrBadSchedule covers both a malformed expression and an unknown timezone. +// Handlers map it to 400 — both are the caller's mistake, and both are much +// cheaper to find at save time than at 2am. +var ErrBadSchedule = errors.New("invalid schedule") + +// Standard 5-field cron: minute hour dom month dow. Deliberately no seconds +// field and no descriptors — a schedule a person cannot read back is a +// schedule nobody can audit. +var cronParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) + +func ParseSchedule(expr, tz string) (cron.Schedule, error) { + if tz == "" { + return nil, fmt.Errorf("%w: a timezone is required", ErrBadSchedule) + } + if _, err := time.LoadLocation(tz); err != nil { + return nil, fmt.Errorf("%w: unknown timezone %q", ErrBadSchedule, tz) + } + sched, err := cronParser.Parse(expr) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrBadSchedule, err) + } + return sched, nil +} + +// NextOccurrence returns the first firing strictly after from, computed in the +// schedule's own zone so that a DST boundary moves the wall-clock time the way +// a person expects rather than drifting by an hour for half the year. +func NextOccurrence(expr, tz string, from time.Time) (time.Time, error) { + sched, err := ParseSchedule(expr, tz) + if err != nil { + return time.Time{}, err + } + loc, err := time.LoadLocation(tz) + if err != nil { + return time.Time{}, fmt.Errorf("%w: unknown timezone %q", ErrBadSchedule, tz) + } + return sched.Next(from.In(loc)), nil +} + +type Decision string + +const ( + Fire Decision = "fire" + SkipMissed Decision = "missed" + SkipRunning Decision = "already_running" +) + +// GraceWindow is how late an occurrence may be and still run. A job missed by +// ten minutes during a deploy should still run; one missed by two days should +// not fire at lunchtime. +const GraceWindow = time.Hour + +// Decide is the whole fire/skip policy, kept pure so it can be tested without +// a database and read without following a loop. +// +// The missed check comes first: an occurrence that is already too old to run +// should be recorded as missed regardless of what is running now, or a slow +// run would relabel a stale occurrence as a fresh conflict. +func Decide(due, now time.Time, runActive bool) Decision { + if now.Sub(due) > GraceWindow { + return SkipMissed + } + if runActive { + return SkipRunning + } + return Fire +}