mrhid6/vantage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity

feat: cron arithmetic and persisted workflow schedules

Browse Source
This commit is contained in:
mrhid6
2026-08-04 13:51:10 +01:00
parent b877024365
commit d0e1cc4ad6
5 changed files with 155 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/go.mod
+5 -2
View File
@@ -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
server/go.sum
+2
View File
@@ -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=
server/internal/models/workflow.go
+18
View File
@@ -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"`
}
server/internal/services/workflows.go
+48
View File
@@ -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
}
server/internal/workflowsched/cron.go
+82
View File
@@ -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
}