feat: fire scheduled workflow runs from the housekeeping leader

2026-08-04 13:53:42 +01:00
parent d0e1cc4ad6
commit a1e6986a64
3 changed files with 213 additions and 0 deletions
@@ -11,6 +11,12 @@ import (
 	"syscall"
 	"time"

+	// Embeds the IANA zone database in the binary. Load-bearing: server/Dockerfile
+	// builds on Alpine, which ships no zoneinfo, so without this
+	// time.LoadLocation("Europe/London") fails in production and every workflow
+	// schedule silently falls back to UTC.
+	_ "time/tzdata"
+
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus"
@@ -18,6 +24,7 @@ import (
 	grpcserver "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched"
 	"github.com/gin-gonic/gin"
 )

@@ -179,6 +186,10 @@ func serve() {
 		services.StartAuditSweeper(jobCtx)
 		services.StartReaper(jobCtx)
 		monitorsched.Start(jobCtx)
+		workflowsched.Start(jobCtx, workflowsched.Deps{
+			TriggerWorkflow: services.TriggerWorkflow,
+			LogEvent:        services.LogEvent,
+		})

 		ticker := time.NewTicker(2 * time.Minute)
 		defer ticker.Stop()
@@ -13,7 +13,9 @@ import (
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/services"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/workflowsched"
 	"github.com/gin-gonic/gin"
+	"go.mongodb.org/mongo-driver/v2/mongo"
 )

 func registerWorkflowRoutes(g *gin.RouterGroup) {
@@ -34,6 +36,8 @@ func registerWorkflowRoutes(g *gin.RouterGroup) {
 	g.DELETE("/workflows/:id", deleteWorkflow)
 	g.POST("/workflows/:id/run", runWorkflow)
 	g.GET("/workflows/:id/runs", listWorkflowRuns)
+	g.PUT("/workflows/:id/schedule", putWorkflowSchedule)
+	g.GET("/workflows/:id/schedule/preview", previewWorkflowSchedule)

 	g.GET("/runs/:runId", getRun)
 	g.POST("/runs/:runId/cancel", cancelRun)
@@ -386,3 +390,51 @@ func cancelRun(c *gin.Context) {
 	services.LogEvent(auth.InstanceID(c), "workflow.run_cancelled", actorFromCtx(c), "", c.Param("runId"), "run cancelled")
 	c.JSON(http.StatusOK, gin.H{"cancelled": true})
 }
+
+func putWorkflowSchedule(c *gin.Context) {
+	var body models.Schedule
+	if err := c.ShouldBindJSON(&body); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid body"})
+		return
+	}
+
+	instanceID := auth.InstanceID(c)
+	next, err := services.SetSchedule(instanceID, c.Param("id"), &body)
+	if err != nil {
+		if errors.Is(err, workflowsched.ErrBadSchedule) {
+			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+			return
+		}
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "workflow not found"})
+			return
+		}
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	services.LogEvent(instanceID, "workflow.schedule_updated", actorFromCtx(c), "", c.Param("id"),
+		fmt.Sprintf("schedule %q %s enabled=%v", body.Cron, body.TZ, body.Enabled))
+	c.JSON(http.StatusOK, gin.H{"schedule": body, "next_run_at": next})
+}
+
+// previewWorkflowSchedule exists so the browser and the scheduler agree on
+// what a cron string means. A client-side cron parser that disagrees with the
+// server by one field is a bug found in production, at night.
+func previewWorkflowSchedule(c *gin.Context) {
+	expr := c.Query("cron")
+	tz := c.Query("tz")
+
+	occurrences := make([]time.Time, 0, 3)
+	from := time.Now()
+	for i := 0; i < 3; i++ {
+		next, err := workflowsched.NextOccurrence(expr, tz, from)
+		if err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+			return
+		}
+		occurrences = append(occurrences, next)
+		from = next
+	}
+	c.JSON(http.StatusOK, gin.H{"occurrences": occurrences})
+}
@@ -0,0 +1,150 @@
+package workflowsched
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+const tickInterval = 30 * time.Second
+
+// Deps are the service functions the loop needs. They are injected rather than
+// imported because services already imports this package for NextOccurrence,
+// and a package cannot import its own importer.
+type Deps struct {
+	TriggerWorkflow func(instanceID, workflowID, actor string) (string, error)
+	LogEvent        func(instanceID, eventType, actor, serverID, keyID, details string)
+}
+
+// Start runs the scheduler until ctx is cancelled. It is called inside
+// bus.RunAsLeader("housekeeping", …) alongside monitorsched and the sweepers:
+// one role, one lock. N replicas each running this loop would fire every
+// scheduled workflow N times.
+func Start(ctx context.Context, deps Deps) {
+	go func() {
+		ticker := time.NewTicker(tickInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				tick(ctx, deps)
+			}
+		}
+	}()
+}
+
+func tick(ctx context.Context, deps Deps) {
+	now := time.Now()
+
+	cur, err := db.Col("workflows").Find(ctx, bson.M{
+		"schedule.enabled": true,
+		"next_run_at":      bson.M{"$lte": now},
+	})
+	if err != nil {
+		log.Printf("workflowsched: find due: %v", err)
+		return
+	}
+	defer cur.Close(ctx)
+
+	var due []models.Workflow
+	if err := cur.All(ctx, &due); err != nil {
+		log.Printf("workflowsched: decode due: %v", err)
+		return
+	}
+
+	for _, wf := range due {
+		if ctx.Err() != nil {
+			return
+		}
+		process(ctx, deps, wf, now)
+	}
+}
+
+func process(ctx context.Context, deps Deps, wf models.Workflow, now time.Time) {
+	if wf.NextRunAt == nil || wf.Schedule == nil {
+		return
+	}
+	dueAt := *wf.NextRunAt
+
+	next, err := NextOccurrence(wf.Schedule.Cron, wf.Schedule.TZ, now)
+	if err != nil {
+		// A schedule that no longer parses cannot be advanced, and leaving
+		// next_run_at in the past would spin this loop every 30 seconds
+		// forever. Disable it and say so.
+		log.Printf("workflowsched: workflow %s has an unusable schedule, disabling: %v", wf.WorkflowID, err)
+		disable(ctx, deps, wf, err.Error())
+		return
+	}
+
+	// The claim. Matching on the current next_run_at as well as the id means a
+	// second process reaching this document after another has claimed it
+	// matches nothing and does nothing. This — not the leader lock — is what
+	// makes a double fire impossible; the lock only keeps it cheap.
+	res, err := db.Col("workflows").UpdateOne(ctx,
+		bson.M{"workflow_id": wf.WorkflowID, "next_run_at": dueAt},
+		bson.M{"$set": bson.M{"next_run_at": next}},
+	)
+	if err != nil {
+		log.Printf("workflowsched: claim %s: %v", wf.WorkflowID, err)
+		return
+	}
+	if res.MatchedCount == 0 {
+		return // claimed elsewhere
+	}
+
+	switch Decide(dueAt, now, hasActiveRun(ctx, wf.InstanceID, wf.WorkflowID)) {
+	case SkipMissed:
+		recordSkip(ctx, deps, wf, string(SkipMissed), dueAt, now)
+	case SkipRunning:
+		recordSkip(ctx, deps, wf, string(SkipRunning), dueAt, now)
+	case Fire:
+		if _, err := deps.TriggerWorkflow(wf.InstanceID, wf.WorkflowID, "schedule"); err != nil {
+			log.Printf("workflowsched: trigger %s: %v", wf.WorkflowID, err)
+			recordSkip(ctx, deps, wf, "error: "+err.Error(), dueAt, now)
+			return
+		}
+		_, _ = db.Col("workflows").UpdateOne(ctx,
+			bson.M{"workflow_id": wf.WorkflowID},
+			bson.M{"$set": bson.M{"last_run_at": now}, "$unset": bson.M{"last_skipped": ""}},
+		)
+		deps.LogEvent(wf.InstanceID, "workflow.scheduled_run", "schedule", "", "",
+			"workflow "+wf.Name+" started on schedule")
+	}
+}
+
+func hasActiveRun(ctx context.Context, instanceID, workflowID string) bool {
+	err := db.Col("workflow_runs").FindOne(ctx, bson.M{
+		"instance_id": instanceID,
+		"workflow_id": workflowID,
+		"status":      "running",
+	}, options.FindOne().SetProjection(bson.M{"_id": 1})).Err()
+	return err == nil
+}
+
+func recordSkip(ctx context.Context, deps Deps, wf models.Workflow, reason string, due, at time.Time) {
+	_, _ = db.Col("workflows").UpdateOne(ctx,
+		bson.M{"workflow_id": wf.WorkflowID},
+		bson.M{"$set": bson.M{"last_skipped": models.Skip{Reason: reason, Due: due, At: at}}},
+	)
+	deps.LogEvent(wf.InstanceID, "workflow.schedule_skipped", "schedule", "", "",
+		"workflow "+wf.Name+" skipped "+due.Format(time.RFC3339)+": "+reason)
+}
+
+func disable(ctx context.Context, deps Deps, wf models.Workflow, reason string) {
+	_, _ = db.Col("workflows").UpdateOne(ctx,
+		bson.M{"workflow_id": wf.WorkflowID},
+		bson.M{
+			"$set":   bson.M{"schedule.enabled": false},
+			"$unset": bson.M{"next_run_at": ""},
+		},
+	)
+	deps.LogEvent(wf.InstanceID, "workflow.schedule_disabled", "schedule", "", "",
+		"workflow "+wf.Name+" schedule disabled: "+reason)
+}