From 3ff24e5fba61ff70daa4c953c231ac82ab1926a2 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 00:29:37 -0700 Subject: [PATCH 01/10] feat(api): add request/response types for cron schedules --- go.mod | 5 ++- go.sum | 2 + internal/api/http/types/cron.go | 76 +++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 1 deletion(-) create mode 100644 internal/api/http/types/cron.go diff --git a/go.mod b/go.mod index 5b80bb9..f1f8b5e 100644 --- a/go.mod +++ b/go.mod @@ -126,4 +126,7 @@ require ( github.com/segmentio/kafka-go v0.4.50 ) -require github.com/pierrec/lz4/v4 v4.1.16 // indirect +require ( + github.com/pierrec/lz4/v4 v4.1.16 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum index d528cb6..be72d79 100644 --- a/go.sum +++ b/go.sum @@ -176,6 +176,8 @@ github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= +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/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= diff --git a/internal/api/http/types/cron.go b/internal/api/http/types/cron.go new file mode 100644 index 0000000..9355e1d --- /dev/null +++ b/internal/api/http/types/cron.go @@ -0,0 +1,76 @@ +package types + +import ( + "encoding/json" + "errors" + "strings" + "time" + + "github.com/robfig/cron/v3" +) + +var ( + ErrInvalidCronExpression = errors.New("invalid cron_expression") + ErrInvalidTimezone = errors.New("invalid timezone") +) + +type CreateCronScheduleRequest struct { + JobDefinitionID string `json:"job_definition_id" binding:"required,uuid"` + CronExpression string `json:"cron_expression" binding:"required"` + Timezone string `json:"timezone" binding:"required"` + PayloadTemplate json.RawMessage `json:"payload_template"` + Description string `json:"description"` +} + +func (r CreateCronScheduleRequest) Validate() error { + expr := strings.TrimSpace(r.CronExpression) + if _, err := cron.ParseStandard(expr); err != nil { + return ErrInvalidCronExpression + } + + tz := strings.TrimSpace(r.Timezone) + if _, err := time.LoadLocation(tz); err != nil { + return ErrInvalidTimezone + } + + return nil +} + +type UpdateCronScheduleRequest struct { + CronExpression *string `json:"cron_expression,omitempty"` + Timezone *string `json:"timezone,omitempty"` + PayloadTemplate *json.RawMessage `json:"payload_template,omitempty"` + Description *string `json:"description,omitempty"` +} + +func (r UpdateCronScheduleRequest) Validate() error { + if r.CronExpression != nil { + expr := strings.TrimSpace(*r.CronExpression) + if _, err := cron.ParseStandard(expr); err != nil { + return ErrInvalidCronExpression + } + } + + if r.Timezone != nil { + tz := strings.TrimSpace(*r.Timezone) + if _, err := time.LoadLocation(tz); err != nil { + return ErrInvalidTimezone + } + } + + return nil +} + +type CronScheduleResponse struct { + ID string `json:"id"` + JobDefinitionID string `json:"job_definition_id"` + CronExpression string `json:"cron_expression"` + Timezone string `json:"timezone"` + PayloadTemplate json.RawMessage `json:"payload_template,omitempty"` + Description string `json:"description,omitempty"` + Status string `json:"status"` + LastFiredAt string `json:"last_fired_at,omitempty"` + NextRunAt string `json:"next_run_at,omitempty"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} From f26156bd2b3571679cb9b84ee83817b537d42745 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 00:40:31 -0700 Subject: [PATCH 02/10] feat(api): implement cron schedules API handlers --- internal/api/http/handlers/cron_handler.go | 226 +++++++++++++++++++++ internal/api/http/router.go | 10 + internal/store/postgres/cron_store.go | 76 ++++++- 3 files changed, 310 insertions(+), 2 deletions(-) create mode 100644 internal/api/http/handlers/cron_handler.go diff --git a/internal/api/http/handlers/cron_handler.go b/internal/api/http/handlers/cron_handler.go new file mode 100644 index 0000000..c4294be --- /dev/null +++ b/internal/api/http/handlers/cron_handler.go @@ -0,0 +1,226 @@ +package handlers + +import ( + "database/sql" + "errors" + "net/http" + "strconv" + "strings" + + "github.com/gin-gonic/gin" + + "github.com/chronos-scheduler/chronos/internal/api/http/response" + apitypes "github.com/chronos-scheduler/chronos/internal/api/http/types" + pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres" +) + +type CronHandler struct { + cronStore *pgstore.CronStore + jobDefStore *pgstore.JobDefinitionStore +} + +func NewCronHandler(cronStore *pgstore.CronStore, jobDefStore *pgstore.JobDefinitionStore) *CronHandler { + return &CronHandler{cronStore: cronStore, jobDefStore: jobDefStore} +} + +func (h *CronHandler) CreateCronSchedule(c *gin.Context) { + var req apitypes.CreateCronScheduleRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.RespondBadRequest(c, "invalid request payload") + return + } + if err := req.Validate(); err != nil { + response.RespondBadRequest(c, err.Error()) + return + } + + if _, err := h.jobDefStore.GetByID(c.Request.Context(), req.JobDefinitionID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "job definition not found") + return + } + response.RespondInternalError(c, "failed to validate job definition") + return + } + + payload := []byte(`{}`) + if len(req.PayloadTemplate) > 0 { + payload = req.PayloadTemplate + } + contextPayload := []byte(`{}`) + + schedule := &pgstore.CronSchedule{ + JobDefID: req.JobDefinitionID, + CronExpression: strings.TrimSpace(req.CronExpression), + Timezone: strings.TrimSpace(req.Timezone), + PayloadOverride: payload, + Context: contextPayload, + Status: "ACTIVE", + } + if err := h.cronStore.Create(c.Request.Context(), schedule); err != nil { + response.RespondInternalError(c, "failed to create cron schedule") + return + } + + c.JSON(http.StatusCreated, toCronScheduleResponse(schedule)) +} + +func (h *CronHandler) ListCronSchedules(c *gin.Context) { + limit := 50 + offset := 0 + if s := strings.TrimSpace(c.Query("limit")); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n <= 0 { + response.RespondBadRequest(c, "invalid limit") + return + } + limit = n + } + if s := strings.TrimSpace(c.Query("offset")); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n < 0 { + response.RespondBadRequest(c, "invalid offset") + return + } + offset = n + } + + status := strings.ToUpper(strings.TrimSpace(c.Query("status"))) + if status != "" && status != "ACTIVE" && status != "PAUSED" && status != "ARCHIVED" { + response.RespondBadRequest(c, "invalid status") + return + } + + schedules, err := h.cronStore.List(c.Request.Context(), pgstore.ListCronSchedulesFilter{ + Status: status, + Limit: limit, + Offset: offset, + }) + if err != nil { + response.RespondInternalError(c, "failed to list cron schedules") + return + } + + items := make([]apitypes.CronScheduleResponse, 0, len(schedules)) + for _, item := range schedules { + items = append(items, toCronScheduleResponse(item)) + } + + c.JSON(http.StatusOK, gin.H{"items": items, "limit": limit, "offset": offset}) +} + +func (h *CronHandler) GetCronSchedule(c *gin.Context) { + schedule, err := h.cronStore.GetByID(c.Request.Context(), c.Param("id")) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to get cron schedule") + return + } + c.JSON(http.StatusOK, toCronScheduleResponse(schedule)) +} + +func (h *CronHandler) UpdateCronSchedule(c *gin.Context) { + schedule, err := h.cronStore.GetByID(c.Request.Context(), c.Param("id")) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to get cron schedule") + return + } + + var req apitypes.UpdateCronScheduleRequest + if err := c.ShouldBindJSON(&req); err != nil { + response.RespondBadRequest(c, "invalid request payload") + return + } + if err := req.Validate(); err != nil { + response.RespondBadRequest(c, err.Error()) + return + } + + if req.CronExpression != nil { + schedule.CronExpression = strings.TrimSpace(*req.CronExpression) + } + if req.Timezone != nil { + schedule.Timezone = strings.TrimSpace(*req.Timezone) + } + if req.PayloadTemplate != nil { + if len(*req.PayloadTemplate) == 0 { + schedule.PayloadOverride = []byte(`{}`) + } else { + schedule.PayloadOverride = *req.PayloadTemplate + } + } + + if err := h.cronStore.Update(c.Request.Context(), schedule); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to update cron schedule") + return + } + + c.JSON(http.StatusOK, toCronScheduleResponse(schedule)) +} + +func (h *CronHandler) PauseCronSchedule(c *gin.Context) { + if err := h.cronStore.SetStatus(c.Request.Context(), c.Param("id"), "PAUSED"); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to pause cron schedule") + return + } + c.Status(http.StatusNoContent) +} + +func (h *CronHandler) ResumeCronSchedule(c *gin.Context) { + if err := h.cronStore.SetStatus(c.Request.Context(), c.Param("id"), "ACTIVE"); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to resume cron schedule") + return + } + c.Status(http.StatusNoContent) +} + +func (h *CronHandler) DeleteCronSchedule(c *gin.Context) { + if err := h.cronStore.SetStatus(c.Request.Context(), c.Param("id"), "ARCHIVED"); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to delete cron schedule") + return + } + c.Status(http.StatusNoContent) +} + +func toCronScheduleResponse(c *pgstore.CronSchedule) apitypes.CronScheduleResponse { + resp := apitypes.CronScheduleResponse{ + ID: c.ID, + JobDefinitionID: c.JobDefID, + CronExpression: c.CronExpression, + Timezone: c.Timezone, + Status: c.Status, + PayloadTemplate: c.PayloadOverride, + CreatedAt: c.CreatedAt.UTC().Format("2006-01-02T15:04:05Z07:00"), + UpdatedAt: c.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z07:00"), + } + if c.LastRunAt != nil { + resp.LastFiredAt = c.LastRunAt.UTC().Format("2006-01-02T15:04:05Z07:00") + } + if c.NextRunAt != nil { + resp.NextRunAt = c.NextRunAt.UTC().Format("2006-01-02T15:04:05Z07:00") + } + return resp +} diff --git a/internal/api/http/router.go b/internal/api/http/router.go index 955d6c0..9d2802f 100644 --- a/internal/api/http/router.go +++ b/internal/api/http/router.go @@ -88,11 +88,13 @@ func NewRouter(deps RouterDeps) *gin.Engine { apiKeyStore := pgstore.NewAPIKeyStore(deps.PostgresDB) jobDefStore := pgstore.NewJobDefinitionStore(deps.PostgresDB) jobStore := pgstore.NewJobStore(deps.PostgresDB) + cronStore := pgstore.NewCronStore(deps.PostgresDB) scheduleBucketStore := pgstore.NewScheduleBucketStore(deps.PostgresDB) jobExecutionStore := pgstore.NewJobExecutionStore(deps.PostgresDB) logStore := mongostore.NewLogStore(deps.MongoClient, "chronos") jobDefHandler := handlers.NewJobDefinitionHandler(jobDefStore) scheduleHandler := handlers.NewScheduleHandler(jobStore, scheduleBucketStore, jobDefStore, jobExecutionStore) + cronHandler := handlers.NewCronHandler(cronStore, jobDefStore) executionHandler := handlers.NewExecutionHandler(jobExecutionStore, logStore) // API v1 @@ -120,6 +122,14 @@ func NewRouter(deps RouterDeps) *gin.Engine { authenticated.GET("/schedules/:id", scheduleHandler.GetSchedule) authenticated.DELETE("/schedules/:id", scheduleHandler.CancelSchedule) + authenticated.POST("/cron-schedules", cronHandler.CreateCronSchedule) + authenticated.GET("/cron-schedules", cronHandler.ListCronSchedules) + authenticated.GET("/cron-schedules/:id", cronHandler.GetCronSchedule) + authenticated.PUT("/cron-schedules/:id", cronHandler.UpdateCronSchedule) + authenticated.POST("/cron-schedules/:id/pause", cronHandler.PauseCronSchedule) + authenticated.POST("/cron-schedules/:id/resume", cronHandler.ResumeCronSchedule) + authenticated.DELETE("/cron-schedules/:id", cronHandler.DeleteCronSchedule) + authenticated.GET("/executions", executionHandler.ListExecutions) authenticated.GET("/executions/:id", executionHandler.GetExecution) } diff --git a/internal/store/postgres/cron_store.go b/internal/store/postgres/cron_store.go index 288ab02..25a7343 100644 --- a/internal/store/postgres/cron_store.go +++ b/internal/store/postgres/cron_store.go @@ -2,6 +2,8 @@ package postgres import ( "context" + "database/sql" + "strconv" "time" "github.com/jmoiron/sqlx" @@ -21,6 +23,12 @@ type CronSchedule struct { UpdatedAt time.Time `db:"updated_at"` } +type ListCronSchedulesFilter struct { + Status string + Limit int + Offset int +} + type CronStore struct { db *sqlx.DB } @@ -45,6 +53,33 @@ func (s *CronStore) Create(ctx context.Context, c *CronSchedule) error { return rows.Err() } +func (s *CronStore) List(ctx context.Context, f ListCronSchedulesFilter) ([]*CronSchedule, error) { + if f.Limit <= 0 { + f.Limit = 50 + } + + query := `SELECT * FROM cron_schedules` + args := []any{} + argIdx := 1 + if f.Status != "" { + query += ` WHERE status = $` + strconv.Itoa(argIdx) + args = append(args, f.Status) + argIdx++ + } + + query += ` ORDER BY created_at DESC LIMIT $` + strconv.Itoa(argIdx) + args = append(args, f.Limit) + argIdx++ + query += ` OFFSET $` + strconv.Itoa(argIdx) + args = append(args, f.Offset) + + var schedules []*CronSchedule + if err := s.db.SelectContext(ctx, &schedules, query, args...); err != nil { + return nil, err + } + return schedules, nil +} + func (s *CronStore) GetByID(ctx context.Context, id string) (*CronSchedule, error) { var c CronSchedule err := s.db.GetContext(ctx, &c, `SELECT * FROM cron_schedules WHERE id = $1`, id) @@ -54,6 +89,33 @@ func (s *CronStore) GetByID(ctx context.Context, id string) (*CronSchedule, erro return &c, nil } +func (s *CronStore) Update(ctx context.Context, c *CronSchedule) error { + res, err := s.db.NamedExecContext(ctx, ` + UPDATE cron_schedules + SET cron_expression = :cron_expression, + timezone = :timezone, + payload_override = :payload_override, + updated_at = NOW() + WHERE id = :id`, c) + if err != nil { + return err + } + rows, err := res.RowsAffected() + if err != nil { + return err + } + if rows == 0 { + return sql.ErrNoRows + } + + updated, err := s.GetByID(ctx, c.ID) + if err != nil { + return err + } + *c = *updated + return nil +} + // ListDue returns active cron schedules whose next_run_at has passed. // Uses FOR UPDATE SKIP LOCKED so concurrent Scheduler replicas don't double-fire. func (s *CronStore) ListDue(ctx context.Context) ([]*CronSchedule, error) { @@ -76,8 +138,18 @@ func (s *CronStore) UpdateNextRunAt(ctx context.Context, id string, lastRunAt, n } func (s *CronStore) SetStatus(ctx context.Context, id, status string) error { - _, err := s.db.ExecContext(ctx, + res, err := s.db.ExecContext(ctx, `UPDATE cron_schedules SET status = $2, updated_at = NOW() WHERE id = $1`, id, status) - return err + if err != nil { + return err + } + rows, err := res.RowsAffected() + if err != nil { + return err + } + if rows == 0 { + return sql.ErrNoRows + } + return nil } From adc9a7553f35fb6d220b649ea8e39e1f4261c426 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 00:49:22 -0700 Subject: [PATCH 03/10] feat(cron): implement next_run_at computation with timezone support and drift prevention --- internal/cron/next_run.go | 51 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 internal/cron/next_run.go diff --git a/internal/cron/next_run.go b/internal/cron/next_run.go new file mode 100644 index 0000000..7e1e07e --- /dev/null +++ b/internal/cron/next_run.go @@ -0,0 +1,51 @@ +package cron + +import ( + "fmt" + "strings" + "time" + + robfigcron "github.com/robfig/cron/v3" +) + +var nowFunc = time.Now + +var standardParser = robfigcron.NewParser( + robfigcron.Minute | + robfigcron.Hour | + robfigcron.Dom | + robfigcron.Month | + robfigcron.Dow, +) + +// NextRun computes the next fire time for a 5-field POSIX cron expression in +// the provided IANA timezone, relative to the current time. +func NextRun(expr, timezone string) (time.Time, error) { + return NextRunFrom(expr, timezone, nowFunc()) +} + +// NextRunFrom computes the next fire time relative to fromTime. +// This is used for drift prevention by passing the scheduled fire time (not the +// actual processing time) as fromTime. +func NextRunFrom(expr, timezone string, fromTime time.Time) (time.Time, error) { + expr = strings.TrimSpace(expr) + if expr == "" { + return time.Time{}, fmt.Errorf("cron expression is required") + } + if strings.TrimSpace(timezone) == "" { + return time.Time{}, fmt.Errorf("timezone is required") + } + + location, err := time.LoadLocation(strings.TrimSpace(timezone)) + if err != nil { + return time.Time{}, fmt.Errorf("load timezone %q: %w", timezone, err) + } + + schedule, err := standardParser.Parse(expr) + if err != nil { + return time.Time{}, fmt.Errorf("parse cron expression %q: %w", expr, err) + } + + next := schedule.Next(fromTime.In(location)) + return next.UTC(), nil +} From cfd3cd87918f1af4c0bcc164a355ad68ea515c8a Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 01:57:21 -0700 Subject: [PATCH 04/10] feat(cron): implement next_run_at computation with timezone support and drift prevention --- cmd/chronos-server/main.go | 43 ++++- internal/cron/scheduler.go | 222 ++++++++++++++++++++++++++ internal/scheduler/dispatcher.go | 4 + internal/store/postgres/cron_store.go | 35 ++++ 4 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 internal/cron/scheduler.go diff --git a/cmd/chronos-server/main.go b/cmd/chronos-server/main.go index 058bf7a..baf1d6d 100644 --- a/cmd/chronos-server/main.go +++ b/cmd/chronos-server/main.go @@ -13,6 +13,7 @@ import ( apihttp "github.com/chronos-scheduler/chronos/internal/api/http" "github.com/chronos-scheduler/chronos/internal/config" + cronengine "github.com/chronos-scheduler/chronos/internal/cron" "github.com/chronos-scheduler/chronos/internal/scheduler" kafkastore "github.com/chronos-scheduler/chronos/internal/store/kafka" mongostore "github.com/chronos-scheduler/chronos/internal/store/mongo" @@ -99,6 +100,12 @@ func main() { cfg.SchedulerBucketSeconds, 0, ) + cronScheduler := cronengine.NewScheduler( + pgstore.NewCronStore(db), + pgstore.NewJobDefinitionStore(db), + pgstore.NewScheduleBucketStore(db), + pgstore.NewJobStore(db), + ) backgroundCtx, cancelBackground := context.WithCancel(context.Background()) defer cancelBackground() @@ -111,16 +118,46 @@ func main() { }() go func() { lockTTL := time.Duration(cfg.LeaderLockTTLSeconds) * time.Second + cronTick := 30 * time.Second logger.Info("scheduler leader loop started", zap.Int("bucket_seconds", cfg.SchedulerBucketSeconds), + zap.Duration("cron_tick", cronTick), zap.Duration("lock_ttl", lockTTL), ) scheduler.RunWithLeadership(backgroundCtx, redisClient, lockTTL, func(leaderCtx context.Context) { logger.Info("scheduler leadership acquired") - if err := dispatcher.Run(leaderCtx); err != nil { - logger.Error("scheduler dispatcher exited with error", zap.Error(err)) + defer logger.Info("scheduler leadership released") + + runOnce := func(now time.Time) error { + if err := cronScheduler.RunOnce(leaderCtx, now); err != nil { + return fmt.Errorf("cron scheduler cycle: %w", err) + } + if err := dispatcher.RunOnce(leaderCtx, now); err != nil { + return fmt.Errorf("bucket dispatcher cycle: %w", err) + } + return nil + } + + if err := runOnce(time.Now().UTC()); err != nil { + logger.Error("scheduler cycle failed", zap.Error(err)) + return + } + + ticker := time.NewTicker(cronTick) + defer ticker.Stop() + + for { + select { + case <-leaderCtx.Done(): + logger.Info("scheduler leader loop stopped", zap.Error(leaderCtx.Err())) + return + case tickAt := <-ticker.C: + if err := runOnce(tickAt.UTC()); err != nil { + logger.Error("scheduler cycle failed", zap.Error(err)) + return + } + } } - logger.Info("scheduler leadership released") }) }() diff --git a/internal/cron/scheduler.go b/internal/cron/scheduler.go new file mode 100644 index 0000000..948284e --- /dev/null +++ b/internal/cron/scheduler.go @@ -0,0 +1,222 @@ +package cron + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/lib/pq" + + pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres" +) + +const defaultInitBatchSize = 200 + +type cronStore interface { + ListActiveWithoutNextRun(ctx context.Context, limit int) ([]*pgstore.CronSchedule, error) + ListDue(ctx context.Context) ([]*pgstore.CronSchedule, error) + SetNextRunAt(ctx context.Context, id string, nextRunAt time.Time) error + UpdateNextRunAt(ctx context.Context, id string, lastRunAt, nextRunAt time.Time) error +} + +type jobDefStore interface { + GetByID(ctx context.Context, id string) (*pgstore.JobDefinition, error) +} + +type bucketStore interface { + Upsert(ctx context.Context, b *pgstore.ScheduleBucket) error +} + +type jobStore interface { + Create(ctx context.Context, j *pgstore.Job) error +} + +type Scheduler struct { + cronStore cronStore + jobDefStore jobDefStore + bucketStore bucketStore + jobStore jobStore + initBatchLimit int +} + +func NewScheduler(cronStore cronStore, jobDefStore jobDefStore, bucketStore bucketStore, jobStore jobStore) *Scheduler { + return &Scheduler{ + cronStore: cronStore, + jobDefStore: jobDefStore, + bucketStore: bucketStore, + jobStore: jobStore, + initBatchLimit: defaultInitBatchSize, + } +} + +// RunOnce processes one cron scheduling cycle: +// 1. Initializes next_run_at for active schedules where next_run_at is NULL. +// 2. Converts due schedules into CRON jobs routed via schedule buckets. +func (s *Scheduler) RunOnce(ctx context.Context, now time.Time) error { + if err := s.initializeNextRuns(ctx); err != nil { + return err + } + if err := s.processDueSchedules(ctx, now.UTC()); err != nil { + return err + } + return nil +} + +func (s *Scheduler) initializeNextRuns(ctx context.Context) error { + for { + items, err := s.cronStore.ListActiveWithoutNextRun(ctx, s.initBatchLimit) + if err != nil { + return err + } + if len(items) == 0 { + return nil + } + + for _, schedule := range items { + nextRunAt, err := NextRun(schedule.CronExpression, schedule.Timezone) + if err != nil { + return fmt.Errorf("compute initial next_run_at for cron %s: %w", schedule.ID, err) + } + if err := s.cronStore.SetNextRunAt(ctx, schedule.ID, nextRunAt); err != nil { + return fmt.Errorf("set initial next_run_at for cron %s: %w", schedule.ID, err) + } + } + + if len(items) < s.initBatchLimit { + return nil + } + } +} + +func (s *Scheduler) processDueSchedules(ctx context.Context, now time.Time) error { + due, err := s.cronStore.ListDue(ctx) + if err != nil { + return err + } + + for _, schedule := range due { + if schedule.NextRunAt == nil { + continue + } + if schedule.NextRunAt.UTC().After(now) { + continue + } + + scheduledFireAt := schedule.NextRunAt.UTC() + if err := s.createCronJob(ctx, schedule, scheduledFireAt); err != nil { + return err + } + + // Drift prevention: compute next run from the scheduled fire time, not actual processing time. + nextRunAt, err := NextRunFrom(schedule.CronExpression, schedule.Timezone, scheduledFireAt) + if err != nil { + return fmt.Errorf("compute next_run_at for cron %s: %w", schedule.ID, err) + } + + if err := s.cronStore.UpdateNextRunAt(ctx, schedule.ID, scheduledFireAt, nextRunAt); err != nil { + return fmt.Errorf("update next_run_at for cron %s: %w", schedule.ID, err) + } + } + + return nil +} + +func (s *Scheduler) createCronJob(ctx context.Context, schedule *pgstore.CronSchedule, scheduledFireAt time.Time) error { + definition, err := s.jobDefStore.GetByID(ctx, schedule.JobDefID) + if err != nil { + return fmt.Errorf("load job definition %s for cron %s: %w", schedule.JobDefID, schedule.ID, err) + } + + payload, err := mergePayloadTemplate(definition.DeliveryConfig, schedule.PayloadOverride) + if err != nil { + return fmt.Errorf("merge payload for cron %s: %w", schedule.ID, err) + } + + bucket := &pgstore.ScheduleBucket{FireAt: scheduledFireAt.Truncate(time.Minute)} + if err := s.bucketStore.Upsert(ctx, bucket); err != nil { + return fmt.Errorf("upsert bucket for cron %s: %w", schedule.ID, err) + } + + cronScheduleID := schedule.ID + jobDefID := definition.ID + job := &pgstore.Job{ + JobType: "CRON", + JobDefID: &jobDefID, + CronScheduleID: &cronScheduleID, + BucketID: &bucket.ID, + DeliveryType: definition.DeliveryType, + DeliveryConfig: definition.DeliveryConfig, + TimeoutSeconds: definition.TimeoutSeconds, + MaxRetries: definition.MaxRetries, + RetryStrategy: definition.RetryStrategy, + RetryBaseDelayS: definition.RetryBaseDelayS, + Payload: payload, + Context: normalizedJSONObject(schedule.Context), + ScheduledAt: ptrTime(scheduledFireAt), + Status: "PENDING", + Tags: pq.StringArray(definition.Tags), + } + if err := s.jobStore.Create(ctx, job); err != nil { + return fmt.Errorf("create cron job for cron %s: %w", schedule.ID, err) + } + + return nil +} + +func mergePayloadTemplate(deliveryConfigJSON, payloadOverrideJSON []byte) ([]byte, error) { + type deliveryConfig struct { + PayloadTemplate json.RawMessage `json:"payload_template"` + } + + cfg := deliveryConfig{} + if len(deliveryConfigJSON) > 0 { + if err := json.Unmarshal(deliveryConfigJSON, &cfg); err != nil { + return nil, fmt.Errorf("decode delivery_config: %w", err) + } + } + + base := map[string]any{} + if len(cfg.PayloadTemplate) > 0 { + if err := json.Unmarshal(cfg.PayloadTemplate, &base); err != nil { + return nil, fmt.Errorf("decode payload_template: %w", err) + } + } + + override := map[string]any{} + if len(payloadOverrideJSON) > 0 { + if err := json.Unmarshal(payloadOverrideJSON, &override); err != nil { + return nil, fmt.Errorf("decode payload_override: %w", err) + } + } + + for k, v := range override { + base[k] = v + } + + return marshalStableJSON(base) +} + +func normalizedJSONObject(raw []byte) []byte { + obj := map[string]any{} + if len(raw) > 0 { + if err := json.Unmarshal(raw, &obj); err == nil { + if b, err := marshalStableJSON(obj); err == nil { + return b + } + } + } + return []byte(`{}`) +} + +func marshalStableJSON(input map[string]any) ([]byte, error) { + if len(input) == 0 { + return []byte(`{}`), nil + } + return json.Marshal(input) +} + +func ptrTime(t time.Time) *time.Time { + v := t + return &v +} diff --git a/internal/scheduler/dispatcher.go b/internal/scheduler/dispatcher.go index 3fa701f..47cc085 100644 --- a/internal/scheduler/dispatcher.go +++ b/internal/scheduler/dispatcher.go @@ -85,6 +85,10 @@ func (d *Dispatcher) Run(ctx context.Context) error { } } +func (d *Dispatcher) RunOnce(ctx context.Context, upTo time.Time) error { + return d.dispatchDueBuckets(ctx, upTo) +} + func (d *Dispatcher) dispatchDueBuckets(ctx context.Context, upTo time.Time) error { buckets, err := d.bucketStore.ListPending(ctx, upTo, defaultBucketScanMax) if err != nil { diff --git a/internal/store/postgres/cron_store.go b/internal/store/postgres/cron_store.go index 25a7343..571bbfa 100644 --- a/internal/store/postgres/cron_store.go +++ b/internal/store/postgres/cron_store.go @@ -80,6 +80,22 @@ func (s *CronStore) List(ctx context.Context, f ListCronSchedulesFilter) ([]*Cro return schedules, nil } +func (s *CronStore) ListActiveWithoutNextRun(ctx context.Context, limit int) ([]*CronSchedule, error) { + if limit <= 0 { + limit = 100 + } + + var schedules []*CronSchedule + err := s.db.SelectContext(ctx, &schedules, + `SELECT * FROM cron_schedules + WHERE status = 'ACTIVE' AND next_run_at IS NULL + ORDER BY created_at ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED`, + limit) + return schedules, err +} + func (s *CronStore) GetByID(ctx context.Context, id string) (*CronSchedule, error) { var c CronSchedule err := s.db.GetContext(ctx, &c, `SELECT * FROM cron_schedules WHERE id = $1`, id) @@ -137,6 +153,25 @@ func (s *CronStore) UpdateNextRunAt(ctx context.Context, id string, lastRunAt, n return err } +func (s *CronStore) SetNextRunAt(ctx context.Context, id string, nextRunAt time.Time) error { + res, err := s.db.ExecContext(ctx, + `UPDATE cron_schedules + SET next_run_at = $2, updated_at = NOW() + WHERE id = $1`, + id, nextRunAt) + if err != nil { + return err + } + rows, err := res.RowsAffected() + if err != nil { + return err + } + if rows == 0 { + return sql.ErrNoRows + } + return nil +} + func (s *CronStore) SetStatus(ctx context.Context, id, status string) error { res, err := s.db.ExecContext(ctx, `UPDATE cron_schedules SET status = $2, updated_at = NOW() WHERE id = $1`, From e52e0f47f1f8b515c0801a07957b1cd8a12b4566 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 04:36:24 -0700 Subject: [PATCH 05/10] feat(cron): add leaderless windowed cron materializer with next_schedule_at cursor --- cmd/chronos-server/main.go | 56 +--- deployments/kind/app/chronos-server.yaml | 4 +- internal/api/http/handlers/cron_handler.go | 4 +- internal/config/config.go | 42 +-- internal/cron/materializer.go | 290 ++++++++++++++++++ internal/cron/payload.go | 58 ++++ internal/cron/scheduler.go | 222 -------------- internal/scheduler/dispatcher.go | 12 +- internal/store/postgres/cron_store.go | 37 +-- internal/store/postgres/store_test.go | 36 ++- ...ext_schedule_at_to_cron_schedules.down.sql | 16 + ..._next_schedule_at_to_cron_schedules.up.sql | 16 + 12 files changed, 459 insertions(+), 334 deletions(-) create mode 100644 internal/cron/materializer.go create mode 100644 internal/cron/payload.go delete mode 100644 internal/cron/scheduler.go create mode 100644 migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.down.sql create mode 100644 migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.up.sql diff --git a/cmd/chronos-server/main.go b/cmd/chronos-server/main.go index baf1d6d..20f0d8c 100644 --- a/cmd/chronos-server/main.go +++ b/cmd/chronos-server/main.go @@ -97,15 +97,12 @@ func main() { pgstore.NewScheduleBucketStore(db), pgstore.NewJobStore(db), kafkaProducer, - cfg.SchedulerBucketSeconds, + cfg.SchedulerBucketTickSeconds, 0, ) - cronScheduler := cronengine.NewScheduler( - pgstore.NewCronStore(db), - pgstore.NewJobDefinitionStore(db), - pgstore.NewScheduleBucketStore(db), - pgstore.NewJobStore(db), - ) + cronTick := time.Duration(cfg.CronMaterializerTickSeconds) * time.Second + cronLookahead := time.Duration(cfg.CronMaterializerLookaheadMinutes) * time.Minute + cronMaterializer := cronengine.NewMaterializer(db, cronTick, cronLookahead, 200, 200) backgroundCtx, cancelBackground := context.WithCancel(context.Background()) defer cancelBackground() @@ -116,47 +113,26 @@ func main() { logger.Fatal("http server error", zap.Error(err)) } }() + go func() { + logger.Info("cron materializer started", + zap.Duration("tick", cronTick), + zap.Duration("lookahead", cronLookahead), + ) + if err := cronMaterializer.Run(backgroundCtx); err != nil { + logger.Error("cron materializer exited with error", zap.Error(err)) + } + }() go func() { lockTTL := time.Duration(cfg.LeaderLockTTLSeconds) * time.Second - cronTick := 30 * time.Second logger.Info("scheduler leader loop started", - zap.Int("bucket_seconds", cfg.SchedulerBucketSeconds), - zap.Duration("cron_tick", cronTick), + zap.Int("bucket_tick_seconds", cfg.SchedulerBucketTickSeconds), zap.Duration("lock_ttl", lockTTL), ) scheduler.RunWithLeadership(backgroundCtx, redisClient, lockTTL, func(leaderCtx context.Context) { logger.Info("scheduler leadership acquired") defer logger.Info("scheduler leadership released") - - runOnce := func(now time.Time) error { - if err := cronScheduler.RunOnce(leaderCtx, now); err != nil { - return fmt.Errorf("cron scheduler cycle: %w", err) - } - if err := dispatcher.RunOnce(leaderCtx, now); err != nil { - return fmt.Errorf("bucket dispatcher cycle: %w", err) - } - return nil - } - - if err := runOnce(time.Now().UTC()); err != nil { - logger.Error("scheduler cycle failed", zap.Error(err)) - return - } - - ticker := time.NewTicker(cronTick) - defer ticker.Stop() - - for { - select { - case <-leaderCtx.Done(): - logger.Info("scheduler leader loop stopped", zap.Error(leaderCtx.Err())) - return - case tickAt := <-ticker.C: - if err := runOnce(tickAt.UTC()); err != nil { - logger.Error("scheduler cycle failed", zap.Error(err)) - return - } - } + if err := dispatcher.Run(leaderCtx); err != nil { + logger.Error("scheduler dispatcher exited with error", zap.Error(err)) } }) }() diff --git a/deployments/kind/app/chronos-server.yaml b/deployments/kind/app/chronos-server.yaml index 1df7201..0a7ec18 100644 --- a/deployments/kind/app/chronos-server.yaml +++ b/deployments/kind/app/chronos-server.yaml @@ -11,7 +11,9 @@ data: CHRONOS_KAFKABROKERS: "chronos-kafka.chronos-infra.svc.cluster.local:9092" CHRONOS_MINIOEENDPOINT: "chronos-minio.chronos-infra.svc.cluster.local:9000" CHRONOS_LEADERLOCKTTLSECONDS: "30" - CHRONOS_SCHEDULERBUCKETSECONDS: "60" + CHRONOS_SCHEDULERBUCKETTICKSECONDS: "60" + CHRONOS_CRONMATERIALIZERTICKSECONDS: "30" + CHRONOS_CRONMATERIALIZERLOOKAHEADMINUTES: "15" --- apiVersion: v1 kind: Secret diff --git a/internal/api/http/handlers/cron_handler.go b/internal/api/http/handlers/cron_handler.go index c4294be..779f1d7 100644 --- a/internal/api/http/handlers/cron_handler.go +++ b/internal/api/http/handlers/cron_handler.go @@ -219,8 +219,8 @@ func toCronScheduleResponse(c *pgstore.CronSchedule) apitypes.CronScheduleRespon if c.LastRunAt != nil { resp.LastFiredAt = c.LastRunAt.UTC().Format("2006-01-02T15:04:05Z07:00") } - if c.NextRunAt != nil { - resp.NextRunAt = c.NextRunAt.UTC().Format("2006-01-02T15:04:05Z07:00") + if c.NextScheduleAt != nil { + resp.NextRunAt = c.NextScheduleAt.UTC().Format("2006-01-02T15:04:05Z07:00") } return resp } diff --git a/internal/config/config.go b/internal/config/config.go index bcbf70b..4b995ad 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -31,7 +31,9 @@ type Config struct { LeaderLockTTLSeconds int // Scheduler - SchedulerBucketSeconds int + SchedulerBucketTickSeconds int + CronMaterializerTickSeconds int + CronMaterializerLookaheadMinutes int // Logging LogLevel string // debug, info, warn, error @@ -57,7 +59,9 @@ func Load() (*Config, error) { v.SetDefault("GRPCPort", 9090) v.SetDefault("CORSAllowedOrigins", []string{"*"}) v.SetDefault("LeaderLockTTLSeconds", 30) - v.SetDefault("SchedulerBucketSeconds", 60) + v.SetDefault("SchedulerBucketTickSeconds", 60) + v.SetDefault("CronMaterializerTickSeconds", 30) + v.SetDefault("CronMaterializerLookaheadMinutes", 15) v.SetDefault("LogLevel", "info") v.SetDefault("LogFormat", "json") v.SetDefault("RedisAddr", "localhost:6379") @@ -74,22 +78,24 @@ func Load() (*Config, error) { } cfg := &Config{ - PostgresDSN: v.GetString("PostgresDSN"), - PostgresMigrationsDir: v.GetString("PostgresMigrationsDir"), - RunMigrations: v.GetBool("RunMigrations"), - MongoURI: v.GetString("MongoURI"), - RedisAddr: v.GetString("RedisAddr"), - KafkaBrokers: v.GetStringSlice("KafkaBrokers"), - MinIOEndpoint: v.GetString("MinIOEndpoint"), - MinIOAccessKey: v.GetString("MinIOAccessKey"), - MinIOSecretKey: v.GetString("MinIOSecretKey"), - HTTPPort: v.GetInt("HTTPPort"), - GRPCPort: v.GetInt("GRPCPort"), - CORSAllowedOrigins: v.GetStringSlice("CORSAllowedOrigins"), - LeaderLockTTLSeconds: v.GetInt("LeaderLockTTLSeconds"), - SchedulerBucketSeconds: v.GetInt("SchedulerBucketSeconds"), - LogLevel: v.GetString("LogLevel"), - LogFormat: v.GetString("LogFormat"), + PostgresDSN: v.GetString("PostgresDSN"), + PostgresMigrationsDir: v.GetString("PostgresMigrationsDir"), + RunMigrations: v.GetBool("RunMigrations"), + MongoURI: v.GetString("MongoURI"), + RedisAddr: v.GetString("RedisAddr"), + KafkaBrokers: v.GetStringSlice("KafkaBrokers"), + MinIOEndpoint: v.GetString("MinIOEndpoint"), + MinIOAccessKey: v.GetString("MinIOAccessKey"), + MinIOSecretKey: v.GetString("MinIOSecretKey"), + HTTPPort: v.GetInt("HTTPPort"), + GRPCPort: v.GetInt("GRPCPort"), + CORSAllowedOrigins: v.GetStringSlice("CORSAllowedOrigins"), + LeaderLockTTLSeconds: v.GetInt("LeaderLockTTLSeconds"), + SchedulerBucketTickSeconds: v.GetInt("SchedulerBucketTickSeconds"), + CronMaterializerTickSeconds: v.GetInt("CronMaterializerTickSeconds"), + CronMaterializerLookaheadMinutes: v.GetInt("CronMaterializerLookaheadMinutes"), + LogLevel: v.GetString("LogLevel"), + LogFormat: v.GetString("LogFormat"), } return cfg, nil diff --git a/internal/cron/materializer.go b/internal/cron/materializer.go new file mode 100644 index 0000000..1f05bed --- /dev/null +++ b/internal/cron/materializer.go @@ -0,0 +1,290 @@ +package cron + +import ( + "context" + "database/sql" + "fmt" + "time" + + "github.com/jmoiron/sqlx" + "github.com/lib/pq" + + pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres" +) + +const ( + defaultTickInterval = 30 * time.Second + defaultLookahead = 15 * time.Minute + defaultClaimBatchMax = 200 + defaultPerScheduleMax = 200 +) + +type Materializer struct { + db *sqlx.DB + tick time.Duration + lookahead time.Duration + claimBatch int + perScheduleMax int + nowProvider func() time.Time +} + +func NewMaterializer(db *sqlx.DB, tick, lookahead time.Duration, claimBatch, perScheduleMax int) *Materializer { + if tick <= 0 { + tick = defaultTickInterval + } + if lookahead <= 0 { + lookahead = defaultLookahead + } + if claimBatch <= 0 { + claimBatch = defaultClaimBatchMax + } + if perScheduleMax <= 0 { + perScheduleMax = defaultPerScheduleMax + } + return &Materializer{ + db: db, + tick: tick, + lookahead: lookahead, + claimBatch: claimBatch, + perScheduleMax: perScheduleMax, + nowProvider: time.Now, + } +} + +func (m *Materializer) Run(ctx context.Context) error { + if err := m.RunOnce(ctx, m.nowProvider().UTC()); err != nil { + return err + } + + ticker := time.NewTicker(m.tick) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return nil + case at := <-ticker.C: + if err := m.RunOnce(ctx, at.UTC()); err != nil { + return err + } + } + } +} + +func (m *Materializer) RunOnce(ctx context.Context, now time.Time) error { + if err := m.initializeCursors(ctx, now); err != nil { + return err + } + if err := m.materializeWindow(ctx, now); err != nil { + return err + } + return nil +} + +func (m *Materializer) initializeCursors(ctx context.Context, now time.Time) error { + for { + tx, err := m.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + + var schedules []*pgstore.CronSchedule + err = tx.SelectContext(ctx, &schedules, + `SELECT * FROM cron_schedules + WHERE status = 'ACTIVE' AND next_schedule_at IS NULL + ORDER BY created_at ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED`, + m.claimBatch) + if err != nil { + _ = tx.Rollback() + return err + } + + if len(schedules) == 0 { + if err := tx.Commit(); err != nil { + return err + } + return nil + } + + for _, schedule := range schedules { + nextScheduleAt, err := NextRunFrom(schedule.CronExpression, schedule.Timezone, now) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("compute next_schedule_at for cron %s: %w", schedule.ID, err) + } + if _, err := tx.ExecContext(ctx, + `UPDATE cron_schedules + SET next_schedule_at = $2, updated_at = NOW() + WHERE id = $1`, + schedule.ID, nextScheduleAt); err != nil { + _ = tx.Rollback() + return fmt.Errorf("set next_schedule_at for cron %s: %w", schedule.ID, err) + } + } + + if err := tx.Commit(); err != nil { + return err + } + + if len(schedules) < m.claimBatch { + return nil + } + } +} + +func (m *Materializer) materializeWindow(ctx context.Context, now time.Time) error { + windowEnd := now.Add(m.lookahead) + + for { + tx, err := m.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + + var schedules []*pgstore.CronSchedule + err = tx.SelectContext(ctx, &schedules, + `SELECT * FROM cron_schedules + WHERE status = 'ACTIVE' AND next_schedule_at IS NOT NULL AND next_schedule_at <= $1 + ORDER BY next_schedule_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED`, + windowEnd, m.claimBatch) + if err != nil { + _ = tx.Rollback() + return err + } + + if len(schedules) == 0 { + if err := tx.Commit(); err != nil { + return err + } + return nil + } + + for _, schedule := range schedules { + if schedule.NextScheduleAt == nil { + continue + } + + definition, err := m.getJobDefinitionTx(ctx, tx, schedule.JobDefID) + if err != nil { + _ = tx.Rollback() + return err + } + payload, err := mergePayloadTemplate(definition.DeliveryConfig, schedule.PayloadOverride) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("merge payload for cron %s: %w", schedule.ID, err) + } + contextPayload := normalizedJSONObject(schedule.Context) + + cursor := schedule.NextScheduleAt.UTC() + lastMaterialized := (*time.Time)(nil) + materializedCount := 0 + for !cursor.After(windowEnd) && materializedCount < m.perScheduleMax { + if err := m.insertCronJobTx(ctx, tx, schedule, definition, payload, contextPayload, cursor); err != nil { + _ = tx.Rollback() + return err + } + materializedCount++ + cm := cursor + lastMaterialized = &cm + nextCursor, err := NextRunFrom(schedule.CronExpression, schedule.Timezone, cursor) + if err != nil { + _ = tx.Rollback() + return fmt.Errorf("advance next_schedule_at for cron %s: %w", schedule.ID, err) + } + cursor = nextCursor + } + + if err := m.advanceScheduleCursorTx(ctx, tx, schedule.ID, cursor, lastMaterialized); err != nil { + _ = tx.Rollback() + return err + } + } + + if err := tx.Commit(); err != nil { + return err + } + } +} + +func (m *Materializer) getJobDefinitionTx(ctx context.Context, tx *sqlx.Tx, id string) (*pgstore.JobDefinition, error) { + var d pgstore.JobDefinition + if err := tx.GetContext(ctx, &d, `SELECT * FROM job_definitions WHERE id = $1 AND deleted_at IS NULL`, id); err != nil { + if err == sql.ErrNoRows { + return nil, fmt.Errorf("job definition %s not found", id) + } + return nil, err + } + return &d, nil +} + +func (m *Materializer) insertCronJobTx(ctx context.Context, tx *sqlx.Tx, schedule *pgstore.CronSchedule, def *pgstore.JobDefinition, payload, contextPayload []byte, scheduledAt time.Time) error { + bucketID, err := upsertBucketTx(ctx, tx, scheduledAt.Truncate(time.Minute)) + if err != nil { + return fmt.Errorf("upsert bucket for cron %s: %w", schedule.ID, err) + } + + _, err = tx.ExecContext(ctx, ` + INSERT INTO jobs ( + job_type, job_def_id, cron_schedule_id, bucket_id, + delivery_type, delivery_config, timeout_seconds, + max_retries, retry_strategy, retry_base_delay_s, + payload, context, scheduled_at, status, tags + ) + VALUES ( + $1, $2, $3, $4, + $5, $6, $7, + $8, $9, $10, + $11, $12, $13, $14, $15 + )`, + "CRON", def.ID, schedule.ID, bucketID, + def.DeliveryType, def.DeliveryConfig, def.TimeoutSeconds, + def.MaxRetries, def.RetryStrategy, def.RetryBaseDelayS, + payload, contextPayload, scheduledAt, "PENDING", pq.StringArray(def.Tags), + ) + if err != nil { + return fmt.Errorf("insert cron job for schedule %s at %s: %w", schedule.ID, scheduledAt.Format(time.RFC3339), err) + } + return nil +} + +func upsertBucketTx(ctx context.Context, tx *sqlx.Tx, fireAt time.Time) (string, error) { + var bucketID string + err := tx.GetContext(ctx, &bucketID, ` + INSERT INTO schedule_buckets (fire_at) + VALUES ($1) + ON CONFLICT (fire_at) DO UPDATE SET fire_at = EXCLUDED.fire_at + RETURNING id`, fireAt) + if err != nil { + return "", err + } + return bucketID, nil +} + +func (m *Materializer) advanceScheduleCursorTx(ctx context.Context, tx *sqlx.Tx, scheduleID string, nextScheduleAt time.Time, lastMaterialized *time.Time) error { + if lastMaterialized != nil { + _, err := tx.ExecContext(ctx, + `UPDATE cron_schedules + SET last_run_at = $2, next_schedule_at = $3, updated_at = NOW() + WHERE id = $1`, + scheduleID, *lastMaterialized, nextScheduleAt) + if err != nil { + return fmt.Errorf("update schedule cursor for cron %s: %w", scheduleID, err) + } + return nil + } + + _, err := tx.ExecContext(ctx, + `UPDATE cron_schedules + SET next_schedule_at = $2, updated_at = NOW() + WHERE id = $1`, + scheduleID, nextScheduleAt) + if err != nil { + return fmt.Errorf("update schedule cursor for cron %s: %w", scheduleID, err) + } + return nil +} diff --git a/internal/cron/payload.go b/internal/cron/payload.go new file mode 100644 index 0000000..60492b1 --- /dev/null +++ b/internal/cron/payload.go @@ -0,0 +1,58 @@ +package cron + +import ( + "encoding/json" + "fmt" +) + +func mergePayloadTemplate(deliveryConfigJSON, payloadOverrideJSON []byte) ([]byte, error) { + type deliveryConfig struct { + PayloadTemplate json.RawMessage `json:"payload_template"` + } + + cfg := deliveryConfig{} + if len(deliveryConfigJSON) > 0 { + if err := json.Unmarshal(deliveryConfigJSON, &cfg); err != nil { + return nil, fmt.Errorf("decode delivery_config: %w", err) + } + } + + base := map[string]any{} + if len(cfg.PayloadTemplate) > 0 { + if err := json.Unmarshal(cfg.PayloadTemplate, &base); err != nil { + return nil, fmt.Errorf("decode payload_template: %w", err) + } + } + + override := map[string]any{} + if len(payloadOverrideJSON) > 0 { + if err := json.Unmarshal(payloadOverrideJSON, &override); err != nil { + return nil, fmt.Errorf("decode payload_override: %w", err) + } + } + + for k, v := range override { + base[k] = v + } + + return marshalJSON(base) +} + +func normalizedJSONObject(raw []byte) []byte { + obj := map[string]any{} + if len(raw) > 0 { + if err := json.Unmarshal(raw, &obj); err == nil { + if b, err := marshalJSON(obj); err == nil { + return b + } + } + } + return []byte(`{}`) +} + +func marshalJSON(input map[string]any) ([]byte, error) { + if len(input) == 0 { + return []byte(`{}`), nil + } + return json.Marshal(input) +} diff --git a/internal/cron/scheduler.go b/internal/cron/scheduler.go deleted file mode 100644 index 948284e..0000000 --- a/internal/cron/scheduler.go +++ /dev/null @@ -1,222 +0,0 @@ -package cron - -import ( - "context" - "encoding/json" - "fmt" - "time" - - "github.com/lib/pq" - - pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres" -) - -const defaultInitBatchSize = 200 - -type cronStore interface { - ListActiveWithoutNextRun(ctx context.Context, limit int) ([]*pgstore.CronSchedule, error) - ListDue(ctx context.Context) ([]*pgstore.CronSchedule, error) - SetNextRunAt(ctx context.Context, id string, nextRunAt time.Time) error - UpdateNextRunAt(ctx context.Context, id string, lastRunAt, nextRunAt time.Time) error -} - -type jobDefStore interface { - GetByID(ctx context.Context, id string) (*pgstore.JobDefinition, error) -} - -type bucketStore interface { - Upsert(ctx context.Context, b *pgstore.ScheduleBucket) error -} - -type jobStore interface { - Create(ctx context.Context, j *pgstore.Job) error -} - -type Scheduler struct { - cronStore cronStore - jobDefStore jobDefStore - bucketStore bucketStore - jobStore jobStore - initBatchLimit int -} - -func NewScheduler(cronStore cronStore, jobDefStore jobDefStore, bucketStore bucketStore, jobStore jobStore) *Scheduler { - return &Scheduler{ - cronStore: cronStore, - jobDefStore: jobDefStore, - bucketStore: bucketStore, - jobStore: jobStore, - initBatchLimit: defaultInitBatchSize, - } -} - -// RunOnce processes one cron scheduling cycle: -// 1. Initializes next_run_at for active schedules where next_run_at is NULL. -// 2. Converts due schedules into CRON jobs routed via schedule buckets. -func (s *Scheduler) RunOnce(ctx context.Context, now time.Time) error { - if err := s.initializeNextRuns(ctx); err != nil { - return err - } - if err := s.processDueSchedules(ctx, now.UTC()); err != nil { - return err - } - return nil -} - -func (s *Scheduler) initializeNextRuns(ctx context.Context) error { - for { - items, err := s.cronStore.ListActiveWithoutNextRun(ctx, s.initBatchLimit) - if err != nil { - return err - } - if len(items) == 0 { - return nil - } - - for _, schedule := range items { - nextRunAt, err := NextRun(schedule.CronExpression, schedule.Timezone) - if err != nil { - return fmt.Errorf("compute initial next_run_at for cron %s: %w", schedule.ID, err) - } - if err := s.cronStore.SetNextRunAt(ctx, schedule.ID, nextRunAt); err != nil { - return fmt.Errorf("set initial next_run_at for cron %s: %w", schedule.ID, err) - } - } - - if len(items) < s.initBatchLimit { - return nil - } - } -} - -func (s *Scheduler) processDueSchedules(ctx context.Context, now time.Time) error { - due, err := s.cronStore.ListDue(ctx) - if err != nil { - return err - } - - for _, schedule := range due { - if schedule.NextRunAt == nil { - continue - } - if schedule.NextRunAt.UTC().After(now) { - continue - } - - scheduledFireAt := schedule.NextRunAt.UTC() - if err := s.createCronJob(ctx, schedule, scheduledFireAt); err != nil { - return err - } - - // Drift prevention: compute next run from the scheduled fire time, not actual processing time. - nextRunAt, err := NextRunFrom(schedule.CronExpression, schedule.Timezone, scheduledFireAt) - if err != nil { - return fmt.Errorf("compute next_run_at for cron %s: %w", schedule.ID, err) - } - - if err := s.cronStore.UpdateNextRunAt(ctx, schedule.ID, scheduledFireAt, nextRunAt); err != nil { - return fmt.Errorf("update next_run_at for cron %s: %w", schedule.ID, err) - } - } - - return nil -} - -func (s *Scheduler) createCronJob(ctx context.Context, schedule *pgstore.CronSchedule, scheduledFireAt time.Time) error { - definition, err := s.jobDefStore.GetByID(ctx, schedule.JobDefID) - if err != nil { - return fmt.Errorf("load job definition %s for cron %s: %w", schedule.JobDefID, schedule.ID, err) - } - - payload, err := mergePayloadTemplate(definition.DeliveryConfig, schedule.PayloadOverride) - if err != nil { - return fmt.Errorf("merge payload for cron %s: %w", schedule.ID, err) - } - - bucket := &pgstore.ScheduleBucket{FireAt: scheduledFireAt.Truncate(time.Minute)} - if err := s.bucketStore.Upsert(ctx, bucket); err != nil { - return fmt.Errorf("upsert bucket for cron %s: %w", schedule.ID, err) - } - - cronScheduleID := schedule.ID - jobDefID := definition.ID - job := &pgstore.Job{ - JobType: "CRON", - JobDefID: &jobDefID, - CronScheduleID: &cronScheduleID, - BucketID: &bucket.ID, - DeliveryType: definition.DeliveryType, - DeliveryConfig: definition.DeliveryConfig, - TimeoutSeconds: definition.TimeoutSeconds, - MaxRetries: definition.MaxRetries, - RetryStrategy: definition.RetryStrategy, - RetryBaseDelayS: definition.RetryBaseDelayS, - Payload: payload, - Context: normalizedJSONObject(schedule.Context), - ScheduledAt: ptrTime(scheduledFireAt), - Status: "PENDING", - Tags: pq.StringArray(definition.Tags), - } - if err := s.jobStore.Create(ctx, job); err != nil { - return fmt.Errorf("create cron job for cron %s: %w", schedule.ID, err) - } - - return nil -} - -func mergePayloadTemplate(deliveryConfigJSON, payloadOverrideJSON []byte) ([]byte, error) { - type deliveryConfig struct { - PayloadTemplate json.RawMessage `json:"payload_template"` - } - - cfg := deliveryConfig{} - if len(deliveryConfigJSON) > 0 { - if err := json.Unmarshal(deliveryConfigJSON, &cfg); err != nil { - return nil, fmt.Errorf("decode delivery_config: %w", err) - } - } - - base := map[string]any{} - if len(cfg.PayloadTemplate) > 0 { - if err := json.Unmarshal(cfg.PayloadTemplate, &base); err != nil { - return nil, fmt.Errorf("decode payload_template: %w", err) - } - } - - override := map[string]any{} - if len(payloadOverrideJSON) > 0 { - if err := json.Unmarshal(payloadOverrideJSON, &override); err != nil { - return nil, fmt.Errorf("decode payload_override: %w", err) - } - } - - for k, v := range override { - base[k] = v - } - - return marshalStableJSON(base) -} - -func normalizedJSONObject(raw []byte) []byte { - obj := map[string]any{} - if len(raw) > 0 { - if err := json.Unmarshal(raw, &obj); err == nil { - if b, err := marshalStableJSON(obj); err == nil { - return b - } - } - } - return []byte(`{}`) -} - -func marshalStableJSON(input map[string]any) ([]byte, error) { - if len(input) == 0 { - return []byte(`{}`), nil - } - return json.Marshal(input) -} - -func ptrTime(t time.Time) *time.Time { - v := t - return &v -} diff --git a/internal/scheduler/dispatcher.go b/internal/scheduler/dispatcher.go index 47cc085..92999c9 100644 --- a/internal/scheduler/dispatcher.go +++ b/internal/scheduler/dispatcher.go @@ -39,7 +39,7 @@ type Dispatcher struct { bucketStore bucketStore jobCounter jobCounter kafkaProducer kafkaPublisher - bucketSeconds int + tickSeconds int shardCount int } @@ -47,11 +47,11 @@ func NewDispatcher( bucketStore bucketStore, jobCounter jobCounter, kafkaProducer kafkaPublisher, - bucketSeconds int, + tickSeconds int, shardCount int, ) *Dispatcher { - if bucketSeconds <= 0 { - bucketSeconds = defaultBucketSeconds + if tickSeconds <= 0 { + tickSeconds = defaultBucketSeconds } if shardCount <= 0 { shardCount = defaultShardCount @@ -60,13 +60,13 @@ func NewDispatcher( bucketStore: bucketStore, jobCounter: jobCounter, kafkaProducer: kafkaProducer, - bucketSeconds: bucketSeconds, + tickSeconds: tickSeconds, shardCount: shardCount, } } func (d *Dispatcher) Run(ctx context.Context) error { - ticker := time.NewTicker(time.Duration(d.bucketSeconds) * time.Second) + ticker := time.NewTicker(time.Duration(d.tickSeconds) * time.Second) defer ticker.Stop() if err := d.dispatchDueBuckets(ctx, time.Now().UTC()); err != nil { diff --git a/internal/store/postgres/cron_store.go b/internal/store/postgres/cron_store.go index 571bbfa..3445a7d 100644 --- a/internal/store/postgres/cron_store.go +++ b/internal/store/postgres/cron_store.go @@ -18,7 +18,7 @@ type CronSchedule struct { Context []byte `db:"context"` // JSONB — runtime vars for delivery_config template substitution Status string `db:"status"` LastRunAt *time.Time `db:"last_run_at"` - NextRunAt *time.Time `db:"next_run_at"` + NextScheduleAt *time.Time `db:"next_schedule_at"` CreatedAt time.Time `db:"created_at"` UpdatedAt time.Time `db:"updated_at"` } @@ -39,8 +39,8 @@ func NewCronStore(db *sqlx.DB) *CronStore { func (s *CronStore) Create(ctx context.Context, c *CronSchedule) error { const q = ` - INSERT INTO cron_schedules (job_def_id, cron_expression, timezone, payload_override, context, status, next_run_at) - VALUES (:job_def_id, :cron_expression, :timezone, :payload_override, :context, :status, :next_run_at) + INSERT INTO cron_schedules (job_def_id, cron_expression, timezone, payload_override, context, status, next_schedule_at) + VALUES (:job_def_id, :cron_expression, :timezone, :payload_override, :context, :status, :next_schedule_at) RETURNING id, created_at, updated_at` rows, err := s.db.NamedQueryContext(ctx, q, c) if err != nil { @@ -80,7 +80,7 @@ func (s *CronStore) List(ctx context.Context, f ListCronSchedulesFilter) ([]*Cro return schedules, nil } -func (s *CronStore) ListActiveWithoutNextRun(ctx context.Context, limit int) ([]*CronSchedule, error) { +func (s *CronStore) ListActiveWithoutNextSchedule(ctx context.Context, limit int) ([]*CronSchedule, error) { if limit <= 0 { limit = 100 } @@ -88,7 +88,7 @@ func (s *CronStore) ListActiveWithoutNextRun(ctx context.Context, limit int) ([] var schedules []*CronSchedule err := s.db.SelectContext(ctx, &schedules, `SELECT * FROM cron_schedules - WHERE status = 'ACTIVE' AND next_run_at IS NULL + WHERE status = 'ACTIVE' AND next_schedule_at IS NULL ORDER BY created_at ASC LIMIT $1 FOR UPDATE SKIP LOCKED`, @@ -132,33 +132,12 @@ func (s *CronStore) Update(ctx context.Context, c *CronSchedule) error { return nil } -// ListDue returns active cron schedules whose next_run_at has passed. -// Uses FOR UPDATE SKIP LOCKED so concurrent Scheduler replicas don't double-fire. -func (s *CronStore) ListDue(ctx context.Context) ([]*CronSchedule, error) { - var schedules []*CronSchedule - err := s.db.SelectContext(ctx, &schedules, - `SELECT * FROM cron_schedules - WHERE status = 'ACTIVE' AND next_run_at <= NOW() - ORDER BY next_run_at ASC - FOR UPDATE SKIP LOCKED`) - return schedules, err -} - -func (s *CronStore) UpdateNextRunAt(ctx context.Context, id string, lastRunAt, nextRunAt time.Time) error { - _, err := s.db.ExecContext(ctx, - `UPDATE cron_schedules - SET last_run_at = $2, next_run_at = $3, updated_at = NOW() - WHERE id = $1`, - id, lastRunAt, nextRunAt) - return err -} - -func (s *CronStore) SetNextRunAt(ctx context.Context, id string, nextRunAt time.Time) error { +func (s *CronStore) SetNextScheduleAt(ctx context.Context, id string, nextScheduleAt time.Time) error { res, err := s.db.ExecContext(ctx, `UPDATE cron_schedules - SET next_run_at = $2, updated_at = NOW() + SET next_schedule_at = $2, updated_at = NOW() WHERE id = $1`, - id, nextRunAt) + id, nextScheduleAt) if err != nil { return err } diff --git a/internal/store/postgres/store_test.go b/internal/store/postgres/store_test.go index ce739d0..a35aed3 100644 --- a/internal/store/postgres/store_test.go +++ b/internal/store/postgres/store_test.go @@ -450,7 +450,7 @@ func TestCronStore(t *testing.T) { PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", - NextRunAt: &nextRun, + NextScheduleAt: &nextRun, } if err := s.Create(ctx, c); err != nil { t.Fatalf("create: %v", err) @@ -464,37 +464,41 @@ func TestCronStore(t *testing.T) { } }) - t.Run("list due", func(t *testing.T) { - pastRun := time.Now().Add(-time.Minute) + t.Run("list active without next schedule", func(t *testing.T) { c := &postgres.CronSchedule{ JobDefID: def.ID, CronExpression: "*/5 * * * *", Timezone: "UTC", - PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", NextRunAt: &pastRun, + PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", } if err := s.Create(ctx, c); err != nil { t.Fatalf("create: %v", err) } - due, err := s.ListDue(ctx) + items, err := s.ListActiveWithoutNextSchedule(ctx, 10) if err != nil { - t.Fatalf("list due: %v", err) + t.Fatalf("list active without next schedule: %v", err) } - if len(due) == 0 { - t.Error("expected at least one due schedule") + if len(items) == 0 { + t.Error("expected at least one schedule without next_schedule_at") } }) - t.Run("update next run at", func(t *testing.T) { - nextRun := time.Now().Add(time.Hour) + t.Run("set next schedule at", func(t *testing.T) { c := &postgres.CronSchedule{ JobDefID: def.ID, CronExpression: "0 0 * * *", Timezone: "UTC", - PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", NextRunAt: &nextRun, + PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", } if err := s.Create(ctx, c); err != nil { t.Fatalf("create: %v", err) } - lastRun := time.Now() - newNextRun := time.Now().Add(24 * time.Hour) - if err := s.UpdateNextRunAt(ctx, c.ID, lastRun, newNextRun); err != nil { - t.Fatalf("update next run at: %v", err) + newNextSchedule := time.Now().Add(24 * time.Hour) + if err := s.SetNextScheduleAt(ctx, c.ID, newNextSchedule); err != nil { + t.Fatalf("set next schedule at: %v", err) + } + got, err := s.GetByID(ctx, c.ID) + if err != nil { + t.Fatalf("get by id: %v", err) + } + if got.NextScheduleAt == nil { + t.Fatal("expected next_schedule_at to be set") } }) @@ -502,7 +506,7 @@ func TestCronStore(t *testing.T) { nextRun := time.Now().Add(time.Hour) c := &postgres.CronSchedule{ JobDefID: def.ID, CronExpression: "0 12 * * *", Timezone: "UTC", - PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", NextRunAt: &nextRun, + PayloadOverride: jsonb(nil), Context: jsonb(nil), Status: "ACTIVE", NextScheduleAt: &nextRun, } if err := s.Create(ctx, c); err != nil { t.Fatalf("create: %v", err) diff --git a/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.down.sql b/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.down.sql new file mode 100644 index 0000000..f5222d3 --- /dev/null +++ b/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.down.sql @@ -0,0 +1,16 @@ +DROP INDEX IF EXISTS idx_cron_schedules_next_schedule; + +ALTER TABLE cron_schedules + ADD COLUMN IF NOT EXISTS next_run_at TIMESTAMPTZ; + +-- Restore next_run_at from the cursor when rolling back. +UPDATE cron_schedules +SET next_run_at = next_schedule_at +WHERE next_run_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_cron_schedules_next_run + ON cron_schedules (next_run_at) + WHERE status = 'ACTIVE'; + +ALTER TABLE cron_schedules + DROP COLUMN IF EXISTS next_schedule_at; diff --git a/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.up.sql b/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.up.sql new file mode 100644 index 0000000..a53719c --- /dev/null +++ b/migrations/postgres/0022_add_next_schedule_at_to_cron_schedules.up.sql @@ -0,0 +1,16 @@ +ALTER TABLE cron_schedules + ADD COLUMN IF NOT EXISTS next_schedule_at TIMESTAMPTZ; + +-- Seed cursor for existing rows before dropping legacy column. +UPDATE cron_schedules +SET next_schedule_at = next_run_at +WHERE next_schedule_at IS NULL AND next_run_at IS NOT NULL; + +DROP INDEX IF EXISTS idx_cron_schedules_next_run; + +ALTER TABLE cron_schedules + DROP COLUMN IF EXISTS next_run_at; + +CREATE INDEX IF NOT EXISTS idx_cron_schedules_next_schedule + ON cron_schedules (next_schedule_at) + WHERE status = 'ACTIVE'; From ec20104ec1d2980fcb2efd9af2727bbed56a7757 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 04:43:59 -0700 Subject: [PATCH 06/10] test(cron): add unit tests for next_schedule_at computation, DST, and drift prevention --- internal/cron/next_run_test.go | 92 ++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 internal/cron/next_run_test.go diff --git a/internal/cron/next_run_test.go b/internal/cron/next_run_test.go new file mode 100644 index 0000000..4ff9900 --- /dev/null +++ b/internal/cron/next_run_test.go @@ -0,0 +1,92 @@ +package cron + +import ( + "testing" + "time" +) + +func TestNextRunFrom_InvalidInputs(t *testing.T) { + _, err := NextRunFrom("", "UTC", time.Now().UTC()) + if err == nil { + t.Fatal("expected error for empty cron expression") + } + + _, err = NextRunFrom("* * * * *", "", time.Now().UTC()) + if err == nil { + t.Fatal("expected error for empty timezone") + } + + _, err = NextRunFrom("invalid", "UTC", time.Now().UTC()) + if err == nil { + t.Fatal("expected error for invalid cron expression") + } + + _, err = NextRunFrom("* * * * *", "Mars/Olympus", time.Now().UTC()) + if err == nil { + t.Fatal("expected error for invalid timezone") + } +} + +func TestNextRun_UsesConfiguredNow(t *testing.T) { + origNow := nowFunc + t.Cleanup(func() { nowFunc = origNow }) + + base := time.Date(2026, 4, 22, 10, 4, 35, 0, time.UTC) + nowFunc = func() time.Time { return base } + + next, err := NextRun("*/5 * * * *", "UTC") + if err != nil { + t.Fatalf("NextRun returned error: %v", err) + } + + want := time.Date(2026, 4, 22, 10, 5, 0, 0, time.UTC) + if !next.Equal(want) { + t.Fatalf("unexpected next run: got %s want %s", next.Format(time.RFC3339), want.Format(time.RFC3339)) + } +} + +func TestNextRunFrom_DSTSpringForward_NewYork(t *testing.T) { + loc, err := time.LoadLocation("America/New_York") + if err != nil { + t.Fatalf("load location: %v", err) + } + + // 2026-03-08 is DST spring-forward day in New York. + fromLocal := time.Date(2026, 3, 8, 1, 59, 0, 0, loc) + next, err := NextRunFrom("30 2 * * *", "America/New_York", fromLocal) + if err != nil { + t.Fatalf("NextRunFrom returned error: %v", err) + } + + nextLocal := next.In(loc) + if nextLocal.Hour() != 2 || nextLocal.Minute() != 30 { + t.Fatalf("expected 02:30 local, got %s", nextLocal.Format(time.RFC3339)) + } + if nextLocal.Day() != 9 { + t.Fatalf("expected next valid run on Mar 9 after spring-forward, got day=%d (%s)", nextLocal.Day(), nextLocal.Format(time.RFC3339)) + } +} + +func TestNextRunFrom_DriftPreventionScheduledVsActual(t *testing.T) { + scheduledFire := time.Date(2026, 4, 22, 10, 0, 0, 0, time.UTC) + actualProcessing := time.Date(2026, 4, 22, 11, 0, 45, 0, time.UTC) + + fromScheduled, err := NextRunFrom("0 * * * *", "UTC", scheduledFire) + if err != nil { + t.Fatalf("NextRunFrom scheduled error: %v", err) + } + fromActual, err := NextRunFrom("0 * * * *", "UTC", actualProcessing) + if err != nil { + t.Fatalf("NextRunFrom actual error: %v", err) + } + + wantScheduled := time.Date(2026, 4, 22, 11, 0, 0, 0, time.UTC) + wantActual := time.Date(2026, 4, 22, 12, 0, 0, 0, time.UTC) + + if !fromScheduled.Equal(wantScheduled) { + t.Fatalf("unexpected next from scheduled: got %s want %s", fromScheduled.Format(time.RFC3339), wantScheduled.Format(time.RFC3339)) + } + if !fromActual.Equal(wantActual) { + t.Fatalf("unexpected next from actual: got %s want %s", fromActual.Format(time.RFC3339), wantActual.Format(time.RFC3339)) + } +} From f2c8a84ba1533a8b5c7d5058df455686f29c2741 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 05:00:32 -0700 Subject: [PATCH 07/10] test(cron): add phase 7 smoke test script --- scripts/smoke-test-phase7.sh | 148 +++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100755 scripts/smoke-test-phase7.sh diff --git a/scripts/smoke-test-phase7.sh b/scripts/smoke-test-phase7.sh new file mode 100755 index 0000000..d89e91d --- /dev/null +++ b/scripts/smoke-test-phase7.sh @@ -0,0 +1,148 @@ +#!/usr/bin/env bash +# Phase 7 smoke test: cron schedule create/pause/resume behavior. +set -euo pipefail + +BASE_URL="${BASE_URL:-http://localhost:8080}" +API_KEY="${API_KEY:-}" +WAIT_FIRST_FIRE_SECONDS="${WAIT_FIRST_FIRE_SECONDS:-180}" +WAIT_PAUSE_VERIFY_SECONDS="${WAIT_PAUSE_VERIFY_SECONDS:-120}" +WAIT_RESUME_VERIFY_SECONDS="${WAIT_RESUME_VERIFY_SECONDS:-180}" +POLL_INTERVAL_SECONDS="${POLL_INTERVAL_SECONDS:-5}" + +if [[ -z "${API_KEY}" ]]; then + echo "API_KEY is required. Seed one with: scripts/seed-api-key.sh" + exit 1 +fi + +json_get() { + local jq_filter="$1" + local py_expr="$2" + if command -v jq >/dev/null 2>&1; then + jq -r "${jq_filter}" + else + python3 -c 'import json,sys; data=json.load(sys.stdin); print(eval(sys.argv[1], {}, {"data": data}))' "${py_expr}" + fi +} + +count_executions() { + local job_def_id="$1" + local resp + resp="$(curl -sS "${BASE_URL}/api/v1/executions?job_def_id=${job_def_id}&limit=200" -H "X-Chronos-API-Key: ${API_KEY}")" + echo "${resp}" | json_get '.items | length' 'len(data.get("items", []))' +} + +NOW_EPOCH="$(date -u +%s)" +RUN_ID="${NOW_EPOCH}" +JOB_NAME="phase7-smoke-${RUN_ID}" + +echo "Running Phase 7 smoke test against ${BASE_URL}..." + +echo "Creating job definition: ${JOB_NAME}" +CREATE_DEF_RESP="$( + curl -sS -X POST "${BASE_URL}/api/v1/job-definitions" \ + -H "X-Chronos-API-Key: ${API_KEY}" \ + -H "Content-Type: application/json" \ + -d "{ + \"name\":\"${JOB_NAME}\", + \"delivery_mode\":\"HTTP\", + \"endpoint\":\"https://httpbin.org/post\", + \"http_method\":\"POST\", + \"timeout_seconds\":15, + \"retry_policy\":{\"max_attempts\":1,\"backoff_array\":[30]} + }" +)" +JOB_DEF_ID="$(echo "${CREATE_DEF_RESP}" | json_get '.id // ""' 'data.get("id", "")')" +if [[ -z "${JOB_DEF_ID}" || "${JOB_DEF_ID}" == "None" ]]; then + echo "Failed to create job definition." + echo "${CREATE_DEF_RESP}" + exit 1 +fi +echo "Job definition created: ${JOB_DEF_ID}" + +echo "Creating cron schedule: * * * * * (UTC)" +CREATE_CRON_RESP="$( + curl -sS -X POST "${BASE_URL}/api/v1/cron-schedules" \ + -H "X-Chronos-API-Key: ${API_KEY}" \ + -H "Content-Type: application/json" \ + -d "{ + \"job_definition_id\":\"${JOB_DEF_ID}\", + \"cron_expression\":\"* * * * *\", + \"timezone\":\"UTC\", + \"payload_template\":{\"phase\":\"7\",\"run_id\":\"${RUN_ID}\"} + }" +)" +CRON_ID="$(echo "${CREATE_CRON_RESP}" | json_get '.id // ""' 'data.get("id", "")')" +if [[ -z "${CRON_ID}" || "${CRON_ID}" == "None" ]]; then + echo "Failed to create cron schedule." + echo "${CREATE_CRON_RESP}" + exit 1 +fi +echo "Cron schedule created: ${CRON_ID}" + +INITIAL_EXEC_COUNT="$(count_executions "${JOB_DEF_ID}")" +echo "Initial execution count: ${INITIAL_EXEC_COUNT}" + +echo "Waiting for first cron fire (timeout ${WAIT_FIRST_FIRE_SECONDS}s)..." +FIRST_DEADLINE=$(( $(date -u +%s) + WAIT_FIRST_FIRE_SECONDS )) +FIRST_FIRED_AT="" +while [[ "$(date -u +%s)" -lt "${FIRST_DEADLINE}" ]]; do + CRON_GET_RESP="$(curl -sS "${BASE_URL}/api/v1/cron-schedules/${CRON_ID}" -H "X-Chronos-API-Key: ${API_KEY}")" + FIRST_FIRED_AT="$(echo "${CRON_GET_RESP}" | json_get '.last_fired_at // ""' 'data.get("last_fired_at", "")')" + if [[ -n "${FIRST_FIRED_AT}" && "${FIRST_FIRED_AT}" != "None" ]]; then + break + fi + sleep "${POLL_INTERVAL_SECONDS}" +done + +if [[ -z "${FIRST_FIRED_AT}" || "${FIRST_FIRED_AT}" == "None" ]]; then + echo "Timed out waiting for cron to fire." + exit 1 +fi +echo "First fire observed at: ${FIRST_FIRED_AT}" + +POST_FIRST_EXEC_COUNT="$(count_executions "${JOB_DEF_ID}")" +echo "Execution count after first fire: ${POST_FIRST_EXEC_COUNT}" +if [[ "${POST_FIRST_EXEC_COUNT}" -le "${INITIAL_EXEC_COUNT}" ]]; then + echo "Expected execution count to increase after first cron fire." + exit 1 +fi + +echo "Pausing cron schedule..." +curl -sS -o /dev/null -X POST "${BASE_URL}/api/v1/cron-schedules/${CRON_ID}/pause" \ + -H "X-Chronos-API-Key: ${API_KEY}" + +EXEC_COUNT_AT_PAUSE="$(count_executions "${JOB_DEF_ID}")" +echo "Execution count at pause: ${EXEC_COUNT_AT_PAUSE}" + +echo "Waiting ${WAIT_PAUSE_VERIFY_SECONDS}s to verify no new executions while paused..." +sleep "${WAIT_PAUSE_VERIFY_SECONDS}" +EXEC_COUNT_AFTER_PAUSE_WAIT="$(count_executions "${JOB_DEF_ID}")" +echo "Execution count after pause wait: ${EXEC_COUNT_AFTER_PAUSE_WAIT}" +if [[ "${EXEC_COUNT_AFTER_PAUSE_WAIT}" -ne "${EXEC_COUNT_AT_PAUSE}" ]]; then + echo "Expected execution count to remain unchanged while cron is paused." + exit 1 +fi + +echo "Resuming cron schedule..." +curl -sS -o /dev/null -X POST "${BASE_URL}/api/v1/cron-schedules/${CRON_ID}/resume" \ + -H "X-Chronos-API-Key: ${API_KEY}" + +echo "Waiting for resumed cron to fire again (timeout ${WAIT_RESUME_VERIFY_SECONDS}s)..." +RESUME_DEADLINE=$(( $(date -u +%s) + WAIT_RESUME_VERIFY_SECONDS )) +RESUMED=false +while [[ "$(date -u +%s)" -lt "${RESUME_DEADLINE}" ]]; do + CURRENT_COUNT="$(count_executions "${JOB_DEF_ID}")" + if [[ "${CURRENT_COUNT}" -gt "${EXEC_COUNT_AFTER_PAUSE_WAIT}" ]]; then + RESUMED=true + echo "Execution count increased after resume: ${CURRENT_COUNT}" + break + fi + sleep "${POLL_INTERVAL_SECONDS}" +done + +if [[ "${RESUMED}" != "true" ]]; then + echo "Timed out waiting for resumed cron to produce a new execution." + exit 1 +fi + +echo "Phase 7 smoke test completed." From 4d2f0fa90f778c5b41cea76c26c43ee3466ae16f Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 05:27:18 -0700 Subject: [PATCH 08/10] feat(scheduler): add tick alignment logic with configurable offset in dispatcher --- internal/scheduler/dispatcher.go | 31 +++++++++++++++++++++++++++---- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/internal/scheduler/dispatcher.go b/internal/scheduler/dispatcher.go index 92999c9..aa8c950 100644 --- a/internal/scheduler/dispatcher.go +++ b/internal/scheduler/dispatcher.go @@ -14,6 +14,7 @@ const ( defaultBucketScanMax = 100 defaultShardCount = 128 targetRowsPerShard = 1000 + dispatchTickOffset = 50 * time.Millisecond ) type BucketTriggerEvent struct { @@ -66,18 +67,21 @@ func NewDispatcher( } func (d *Dispatcher) Run(ctx context.Context) error { - ticker := time.NewTicker(time.Duration(d.tickSeconds) * time.Second) - defer ticker.Stop() - if err := d.dispatchDueBuckets(ctx, time.Now().UTC()); err != nil { return err } + interval := time.Duration(d.tickSeconds) * time.Second for { + nextTick := nextAlignedTick(time.Now().UTC(), interval, dispatchTickOffset) + timer := time.NewTimer(time.Until(nextTick)) select { case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } return nil - case <-ticker.C: + case <-timer.C: if err := d.dispatchDueBuckets(ctx, time.Now().UTC()); err != nil { return err } @@ -85,6 +89,25 @@ func (d *Dispatcher) Run(ctx context.Context) error { } } +func nextAlignedTick(now time.Time, interval, offset time.Duration) time.Time { + if interval <= 0 { + interval = time.Duration(defaultBucketSeconds) * time.Second + } + if offset < 0 { + offset = 0 + } + if offset >= interval { + offset = offset % interval + } + + base := now.Truncate(interval) + candidate := base.Add(offset) + if !candidate.After(now) { + candidate = base.Add(interval).Add(offset) + } + return candidate +} + func (d *Dispatcher) RunOnce(ctx context.Context, upTo time.Time) error { return d.dispatchDueBuckets(ctx, upTo) } From 9c6e321612d98a49e16355bedce1a1de059ca52b Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 05:40:04 -0700 Subject: [PATCH 09/10] feat(scheduler): add fireLookahead to dispatcher for improved scheduling accuracy --- internal/scheduler/dispatcher.go | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/internal/scheduler/dispatcher.go b/internal/scheduler/dispatcher.go index aa8c950..8a24eb5 100644 --- a/internal/scheduler/dispatcher.go +++ b/internal/scheduler/dispatcher.go @@ -14,7 +14,7 @@ const ( defaultBucketScanMax = 100 defaultShardCount = 128 targetRowsPerShard = 1000 - dispatchTickOffset = 50 * time.Millisecond + defaultFireLookahead = 1 * time.Second ) type BucketTriggerEvent struct { @@ -42,6 +42,7 @@ type Dispatcher struct { kafkaProducer kafkaPublisher tickSeconds int shardCount int + fireLookahead time.Duration } func NewDispatcher( @@ -63,17 +64,18 @@ func NewDispatcher( kafkaProducer: kafkaProducer, tickSeconds: tickSeconds, shardCount: shardCount, + fireLookahead: defaultFireLookahead, } } func (d *Dispatcher) Run(ctx context.Context) error { - if err := d.dispatchDueBuckets(ctx, time.Now().UTC()); err != nil { + if err := d.dispatchDueBuckets(ctx, time.Now().UTC().Add(d.fireLookahead)); err != nil { return err } interval := time.Duration(d.tickSeconds) * time.Second for { - nextTick := nextAlignedTick(time.Now().UTC(), interval, dispatchTickOffset) + nextTick := nextAlignedTick(time.Now().UTC(), interval) timer := time.NewTimer(time.Until(nextTick)) select { case <-ctx.Done(): @@ -82,28 +84,20 @@ func (d *Dispatcher) Run(ctx context.Context) error { } return nil case <-timer.C: - if err := d.dispatchDueBuckets(ctx, time.Now().UTC()); err != nil { + if err := d.dispatchDueBuckets(ctx, time.Now().UTC().Add(d.fireLookahead)); err != nil { return err } } } } -func nextAlignedTick(now time.Time, interval, offset time.Duration) time.Time { +func nextAlignedTick(now time.Time, interval time.Duration) time.Time { if interval <= 0 { interval = time.Duration(defaultBucketSeconds) * time.Second } - if offset < 0 { - offset = 0 - } - if offset >= interval { - offset = offset % interval - } - - base := now.Truncate(interval) - candidate := base.Add(offset) + candidate := now.Truncate(interval).Add(interval) if !candidate.After(now) { - candidate = base.Add(interval).Add(offset) + candidate = candidate.Add(interval) } return candidate } From 491d94f39337cf8d8bba6e68676d69e486f865a5 Mon Sep 17 00:00:00 2001 From: chandan-m Date: Wed, 22 Apr 2026 06:13:01 -0700 Subject: [PATCH 10/10] feat(cron): add reconciliation, history endpoints, and enhanced materializer integration --- cmd/chronos-server/main.go | 8 +- internal/api/http/handlers/cron_handler.go | 141 ++++++++++++++++++++- internal/api/http/router.go | 7 +- internal/api/http/types/cron.go | 14 ++ internal/cron/materializer.go | 73 +++++++++++ internal/cron/payload_test.go | 64 ++++++++++ internal/store/postgres/cron_store.go | 51 ++++++++ internal/store/postgres/job_store.go | 16 +++ scripts/port-forward.sh | 75 +++++++---- scripts/smoke-test-phase7.sh | 11 +- 10 files changed, 427 insertions(+), 33 deletions(-) create mode 100644 internal/cron/payload_test.go diff --git a/cmd/chronos-server/main.go b/cmd/chronos-server/main.go index 20f0d8c..d90484a 100644 --- a/cmd/chronos-server/main.go +++ b/cmd/chronos-server/main.go @@ -80,12 +80,17 @@ func main() { } }() + cronTick := time.Duration(cfg.CronMaterializerTickSeconds) * time.Second + cronLookahead := time.Duration(cfg.CronMaterializerLookaheadMinutes) * time.Minute + cronMaterializer := cronengine.NewMaterializer(db, cronTick, cronLookahead, 200, 200) + router := apihttp.NewRouter(apihttp.RouterDeps{ Logger: logger, PostgresDB: db, MongoClient: mongoClient, RedisClient: redisClient, CORSAllowedOrigins: cfg.CORSAllowedOrigins, + CronReconciler: cronMaterializer, }) srv := &http.Server{ @@ -100,9 +105,6 @@ func main() { cfg.SchedulerBucketTickSeconds, 0, ) - cronTick := time.Duration(cfg.CronMaterializerTickSeconds) * time.Second - cronLookahead := time.Duration(cfg.CronMaterializerLookaheadMinutes) * time.Minute - cronMaterializer := cronengine.NewMaterializer(db, cronTick, cronLookahead, 200, 200) backgroundCtx, cancelBackground := context.WithCancel(context.Background()) defer cancelBackground() diff --git a/internal/api/http/handlers/cron_handler.go b/internal/api/http/handlers/cron_handler.go index 779f1d7..b0d35f0 100644 --- a/internal/api/http/handlers/cron_handler.go +++ b/internal/api/http/handlers/cron_handler.go @@ -1,26 +1,40 @@ package handlers import ( + "context" "database/sql" "errors" "net/http" "strconv" "strings" + "time" "github.com/gin-gonic/gin" "github.com/chronos-scheduler/chronos/internal/api/http/response" apitypes "github.com/chronos-scheduler/chronos/internal/api/http/types" + cronengine "github.com/chronos-scheduler/chronos/internal/cron" pgstore "github.com/chronos-scheduler/chronos/internal/store/postgres" ) +type cronReconciler interface { + ReconcileSchedule(ctx context.Context, scheduleID string, now time.Time) error +} + type CronHandler struct { cronStore *pgstore.CronStore jobDefStore *pgstore.JobDefinitionStore + jobStore *pgstore.JobStore + reconciler cronReconciler } -func NewCronHandler(cronStore *pgstore.CronStore, jobDefStore *pgstore.JobDefinitionStore) *CronHandler { - return &CronHandler{cronStore: cronStore, jobDefStore: jobDefStore} +func NewCronHandler(cronStore *pgstore.CronStore, jobDefStore *pgstore.JobDefinitionStore, jobStore *pgstore.JobStore, reconciler cronReconciler) *CronHandler { + return &CronHandler{ + cronStore: cronStore, + jobDefStore: jobDefStore, + jobStore: jobStore, + reconciler: reconciler, + } } func (h *CronHandler) CreateCronSchedule(c *gin.Context) { @@ -122,6 +136,49 @@ func (h *CronHandler) GetCronSchedule(c *gin.Context) { c.JSON(http.StatusOK, toCronScheduleResponse(schedule)) } +func (h *CronHandler) GetCronScheduleHistory(c *gin.Context) { + limit := 50 + offset := 0 + if s := strings.TrimSpace(c.Query("limit")); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n <= 0 { + response.RespondBadRequest(c, "invalid limit") + return + } + limit = n + } + if s := strings.TrimSpace(c.Query("offset")); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n < 0 { + response.RespondBadRequest(c, "invalid offset") + return + } + offset = n + } + + cronID := c.Param("id") + if _, err := h.cronStore.GetByID(c.Request.Context(), cronID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to get cron schedule") + return + } + + rows, err := h.cronStore.ListHistory(c.Request.Context(), cronID, limit, offset) + if err != nil { + response.RespondInternalError(c, "failed to list cron schedule history") + return + } + + items := make([]apitypes.CronScheduleHistoryItem, 0, len(rows)) + for _, row := range rows { + items = append(items, toCronScheduleHistoryItem(row)) + } + c.JSON(http.StatusOK, gin.H{"items": items, "limit": limit, "offset": offset}) +} + func (h *CronHandler) UpdateCronSchedule(c *gin.Context) { schedule, err := h.cronStore.GetByID(c.Request.Context(), c.Param("id")) if err != nil { @@ -166,6 +223,27 @@ func (h *CronHandler) UpdateCronSchedule(c *gin.Context) { return } + now := time.Now().UTC() + nextScheduleAt, err := cronengine.NextRun(schedule.CronExpression, schedule.Timezone) + if err != nil { + response.RespondInternalError(c, "failed to compute next schedule time") + return + } + if err := h.cronStore.SetNextScheduleAt(c.Request.Context(), schedule.ID, nextScheduleAt); err != nil { + response.RespondInternalError(c, "failed to reset next schedule time") + return + } + if _, err := h.jobStore.CancelPendingByCronSchedule(c.Request.Context(), schedule.ID); err != nil { + response.RespondInternalError(c, "failed to cancel pending cron jobs during update") + return + } + if h.reconciler != nil && schedule.Status == "ACTIVE" { + if err := h.reconciler.ReconcileSchedule(c.Request.Context(), schedule.ID, now); err != nil { + response.RespondInternalError(c, "failed to rematerialize cron jobs after update") + return + } + } + c.JSON(http.StatusOK, toCronScheduleResponse(schedule)) } @@ -178,10 +256,23 @@ func (h *CronHandler) PauseCronSchedule(c *gin.Context) { response.RespondInternalError(c, "failed to pause cron schedule") return } + if _, err := h.jobStore.CancelPendingByCronSchedule(c.Request.Context(), c.Param("id")); err != nil { + response.RespondInternalError(c, "failed to cancel pending cron jobs on pause") + return + } c.Status(http.StatusNoContent) } func (h *CronHandler) ResumeCronSchedule(c *gin.Context) { + schedule, err := h.cronStore.GetByID(c.Request.Context(), c.Param("id")) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + response.RespondNotFound(c, "cron schedule not found") + return + } + response.RespondInternalError(c, "failed to get cron schedule") + return + } if err := h.cronStore.SetStatus(c.Request.Context(), c.Param("id"), "ACTIVE"); err != nil { if errors.Is(err, sql.ErrNoRows) { response.RespondNotFound(c, "cron schedule not found") @@ -190,6 +281,21 @@ func (h *CronHandler) ResumeCronSchedule(c *gin.Context) { response.RespondInternalError(c, "failed to resume cron schedule") return } + nextScheduleAt, err := cronengine.NextRun(schedule.CronExpression, schedule.Timezone) + if err != nil { + response.RespondInternalError(c, "failed to compute next schedule time on resume") + return + } + if err := h.cronStore.SetNextScheduleAt(c.Request.Context(), schedule.ID, nextScheduleAt); err != nil { + response.RespondInternalError(c, "failed to set next schedule time on resume") + return + } + if h.reconciler != nil { + if err := h.reconciler.ReconcileSchedule(c.Request.Context(), schedule.ID, time.Now().UTC()); err != nil { + response.RespondInternalError(c, "failed to rematerialize cron jobs on resume") + return + } + } c.Status(http.StatusNoContent) } @@ -224,3 +330,34 @@ func toCronScheduleResponse(c *pgstore.CronSchedule) apitypes.CronScheduleRespon } return resp } + +func toCronScheduleHistoryItem(row *pgstore.CronScheduleHistoryRow) apitypes.CronScheduleHistoryItem { + item := apitypes.CronScheduleHistoryItem{ + JobID: row.JobID, + JobStatus: row.JobStatus, + CreatedAt: row.CreatedAt.UTC().Format("2006-01-02T15:04:05Z07:00"), + } + if row.ScheduledAt != nil { + item.ScheduledAt = row.ScheduledAt.UTC().Format("2006-01-02T15:04:05Z07:00") + } + if row.ExecutionID != nil { + item.ExecutionID = *row.ExecutionID + } + if row.AttemptNumber != nil { + item.AttemptNumber = *row.AttemptNumber + } + if row.ExecutionStatus != nil { + item.ExecutionStatus = *row.ExecutionStatus + } + if row.ExecutionStartedAt != nil { + item.ExecutionStarted = row.ExecutionStartedAt.UTC().Format("2006-01-02T15:04:05Z07:00") + } + if row.ExecutionEndedAt != nil { + item.ExecutionEnded = row.ExecutionEndedAt.UTC().Format("2006-01-02T15:04:05Z07:00") + } + item.ResponseStatus = row.ResponseStatus + if row.ErrorMessage != nil { + item.ErrorMessage = *row.ErrorMessage + } + return item +} diff --git a/internal/api/http/router.go b/internal/api/http/router.go index 9d2802f..61e1363 100644 --- a/internal/api/http/router.go +++ b/internal/api/http/router.go @@ -28,6 +28,9 @@ type RouterDeps struct { MongoClient *mongo.Client RedisClient *redis.Client CORSAllowedOrigins []string + CronReconciler interface { + ReconcileSchedule(ctx context.Context, scheduleID string, now time.Time) error + } } func NewRouter(deps RouterDeps) *gin.Engine { @@ -94,7 +97,7 @@ func NewRouter(deps RouterDeps) *gin.Engine { logStore := mongostore.NewLogStore(deps.MongoClient, "chronos") jobDefHandler := handlers.NewJobDefinitionHandler(jobDefStore) scheduleHandler := handlers.NewScheduleHandler(jobStore, scheduleBucketStore, jobDefStore, jobExecutionStore) - cronHandler := handlers.NewCronHandler(cronStore, jobDefStore) + cronHandler := handlers.NewCronHandler(cronStore, jobDefStore, jobStore, deps.CronReconciler) executionHandler := handlers.NewExecutionHandler(jobExecutionStore, logStore) // API v1 @@ -125,6 +128,8 @@ func NewRouter(deps RouterDeps) *gin.Engine { authenticated.POST("/cron-schedules", cronHandler.CreateCronSchedule) authenticated.GET("/cron-schedules", cronHandler.ListCronSchedules) authenticated.GET("/cron-schedules/:id", cronHandler.GetCronSchedule) + authenticated.GET("/cron-schedules/:id/history", cronHandler.GetCronScheduleHistory) + authenticated.GET("/cron-schedule/:id/history", cronHandler.GetCronScheduleHistory) authenticated.PUT("/cron-schedules/:id", cronHandler.UpdateCronSchedule) authenticated.POST("/cron-schedules/:id/pause", cronHandler.PauseCronSchedule) authenticated.POST("/cron-schedules/:id/resume", cronHandler.ResumeCronSchedule) diff --git a/internal/api/http/types/cron.go b/internal/api/http/types/cron.go index 9355e1d..8b25672 100644 --- a/internal/api/http/types/cron.go +++ b/internal/api/http/types/cron.go @@ -74,3 +74,17 @@ type CronScheduleResponse struct { CreatedAt string `json:"created_at"` UpdatedAt string `json:"updated_at"` } + +type CronScheduleHistoryItem struct { + JobID string `json:"job_id"` + ScheduledAt string `json:"scheduled_at,omitempty"` + JobStatus string `json:"job_status"` + CreatedAt string `json:"created_at"` + ExecutionID string `json:"execution_id,omitempty"` + AttemptNumber int `json:"attempt_number,omitempty"` + ExecutionStatus string `json:"execution_status,omitempty"` + ExecutionStarted string `json:"execution_started_at,omitempty"` + ExecutionEnded string `json:"execution_completed_at,omitempty"` + ResponseStatus *int `json:"response_status,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` +} diff --git a/internal/cron/materializer.go b/internal/cron/materializer.go index 1f05bed..3c39666 100644 --- a/internal/cron/materializer.go +++ b/internal/cron/materializer.go @@ -81,6 +81,79 @@ func (m *Materializer) RunOnce(ctx context.Context, now time.Time) error { return nil } +// ReconcileSchedule rematerializes a single schedule into the current lookahead +// window. Intended for immediate reconciliation on pause/resume/update flows. +func (m *Materializer) ReconcileSchedule(ctx context.Context, scheduleID string, now time.Time) error { + windowEnd := now.Add(m.lookahead) + return m.materializeSingleScheduleTx(ctx, scheduleID, windowEnd) +} + +func (m *Materializer) materializeSingleScheduleTx(ctx context.Context, scheduleID string, windowEnd time.Time) error { + tx, err := m.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { + if tx != nil { + _ = tx.Rollback() + } + }() + + var schedule pgstore.CronSchedule + if err := tx.GetContext(ctx, &schedule, ` + SELECT * FROM cron_schedules + WHERE id = $1 + FOR UPDATE`, scheduleID); err != nil { + if err == sql.ErrNoRows { + return nil + } + return err + } + if schedule.Status != "ACTIVE" || schedule.NextScheduleAt == nil { + if err := tx.Commit(); err != nil { + return err + } + tx = nil + return nil + } + + definition, err := m.getJobDefinitionTx(ctx, tx, schedule.JobDefID) + if err != nil { + return err + } + payload, err := mergePayloadTemplate(definition.DeliveryConfig, schedule.PayloadOverride) + if err != nil { + return fmt.Errorf("merge payload for cron %s: %w", schedule.ID, err) + } + contextPayload := normalizedJSONObject(schedule.Context) + + cursor := schedule.NextScheduleAt.UTC() + var lastMaterialized *time.Time + materializedCount := 0 + for !cursor.After(windowEnd) && materializedCount < m.perScheduleMax { + if err := m.insertCronJobTx(ctx, tx, &schedule, definition, payload, contextPayload, cursor); err != nil { + return err + } + materializedCount++ + cm := cursor + lastMaterialized = &cm + nextCursor, err := NextRunFrom(schedule.CronExpression, schedule.Timezone, cursor) + if err != nil { + return fmt.Errorf("advance next_schedule_at for cron %s: %w", schedule.ID, err) + } + cursor = nextCursor + } + if err := m.advanceScheduleCursorTx(ctx, tx, schedule.ID, cursor, lastMaterialized); err != nil { + return err + } + + if err := tx.Commit(); err != nil { + return err + } + tx = nil + return nil +} + func (m *Materializer) initializeCursors(ctx context.Context, now time.Time) error { for { tx, err := m.db.BeginTxx(ctx, nil) diff --git a/internal/cron/payload_test.go b/internal/cron/payload_test.go new file mode 100644 index 0000000..2785dc5 --- /dev/null +++ b/internal/cron/payload_test.go @@ -0,0 +1,64 @@ +package cron + +import ( + "encoding/json" + "testing" +) + +func TestMergePayloadTemplate_OverrideWins(t *testing.T) { + deliveryConfig := []byte(`{"payload_template":{"a":1,"b":"base","nested":{"x":1}}}`) + override := []byte(`{"b":"override","c":true}`) + + out, err := mergePayloadTemplate(deliveryConfig, override) + if err != nil { + t.Fatalf("mergePayloadTemplate error: %v", err) + } + + var got map[string]any + if err := json.Unmarshal(out, &got); err != nil { + t.Fatalf("unmarshal output: %v", err) + } + + if got["a"].(float64) != 1 { + t.Fatalf("expected a=1, got %v", got["a"]) + } + if got["b"].(string) != "override" { + t.Fatalf("expected b=override, got %v", got["b"]) + } + if got["c"].(bool) != true { + t.Fatalf("expected c=true, got %v", got["c"]) + } +} + +func TestMergePayloadTemplate_InvalidJSON(t *testing.T) { + _, err := mergePayloadTemplate([]byte(`{"payload_template":`), []byte(`{}`)) + if err == nil { + t.Fatal("expected error for invalid delivery_config JSON") + } + + _, err = mergePayloadTemplate([]byte(`{"payload_template":{}}`), []byte(`{"x":`)) + if err == nil { + t.Fatal("expected error for invalid payload_override JSON") + } +} + +func TestNormalizedJSONObject_FallbackAndNormalize(t *testing.T) { + if got := string(normalizedJSONObject(nil)); got != "{}" { + t.Fatalf("expected {}, got %s", got) + } + if got := string(normalizedJSONObject([]byte("invalid"))); got != "{}" { + t.Fatalf("expected {} for invalid input, got %s", got) + } + + got := normalizedJSONObject([]byte(`{"tenant":"acme","k":1}`)) + var obj map[string]any + if err := json.Unmarshal(got, &obj); err != nil { + t.Fatalf("unmarshal normalized object: %v", err) + } + if obj["tenant"].(string) != "acme" { + t.Fatalf("expected tenant=acme, got %v", obj["tenant"]) + } + if obj["k"].(float64) != 1 { + t.Fatalf("expected k=1, got %v", obj["k"]) + } +} diff --git a/internal/store/postgres/cron_store.go b/internal/store/postgres/cron_store.go index 3445a7d..2f9de6a 100644 --- a/internal/store/postgres/cron_store.go +++ b/internal/store/postgres/cron_store.go @@ -23,6 +23,20 @@ type CronSchedule struct { UpdatedAt time.Time `db:"updated_at"` } +type CronScheduleHistoryRow struct { + JobID string `db:"job_id"` + ScheduledAt *time.Time `db:"scheduled_at"` + JobStatus string `db:"job_status"` + CreatedAt time.Time `db:"created_at"` + ExecutionID *string `db:"execution_id"` + AttemptNumber *int `db:"attempt_number"` + ExecutionStatus *string `db:"execution_status"` + ExecutionStartedAt *time.Time `db:"execution_started_at"` + ExecutionEndedAt *time.Time `db:"execution_completed_at"` + ResponseStatus *int `db:"response_status"` + ErrorMessage *string `db:"error_message"` +} + type ListCronSchedulesFilter struct { Status string Limit int @@ -167,3 +181,40 @@ func (s *CronStore) SetStatus(ctx context.Context, id, status string) error { } return nil } + +func (s *CronStore) ListHistory(ctx context.Context, cronScheduleID string, limit, offset int) ([]*CronScheduleHistoryRow, error) { + if limit <= 0 { + limit = 50 + } + + var rows []*CronScheduleHistoryRow + err := s.db.SelectContext(ctx, &rows, ` + SELECT + j.id AS job_id, + j.scheduled_at AS scheduled_at, + j.status AS job_status, + j.created_at AS created_at, + je.id AS execution_id, + je.attempt_number AS attempt_number, + je.status AS execution_status, + je.started_at AS execution_started_at, + je.completed_at AS execution_completed_at, + je.response_status AS response_status, + je.error_message AS error_message + FROM jobs j + LEFT JOIN LATERAL ( + SELECT * + FROM job_executions je + WHERE je.job_id = j.id + ORDER BY je.attempt_number DESC + LIMIT 1 + ) je ON TRUE + WHERE j.cron_schedule_id = $1 + ORDER BY j.scheduled_at DESC NULLS LAST, j.created_at DESC + LIMIT $2 OFFSET $3`, + cronScheduleID, limit, offset) + if err != nil { + return nil, err + } + return rows, nil +} diff --git a/internal/store/postgres/job_store.go b/internal/store/postgres/job_store.go index dfc4449..012aad5 100644 --- a/internal/store/postgres/job_store.go +++ b/internal/store/postgres/job_store.go @@ -273,3 +273,19 @@ func (s *JobStore) Cancel(ctx context.Context, id string) error { } return nil } + +func (s *JobStore) CancelPendingByCronSchedule(ctx context.Context, cronScheduleID string) (int64, error) { + res, err := s.db.ExecContext(ctx, + `UPDATE jobs + SET status = 'CANCELLED' + WHERE cron_schedule_id = $1 AND status = 'PENDING'`, + cronScheduleID) + if err != nil { + return 0, err + } + rowsAffected, err := res.RowsAffected() + if err != nil { + return 0, err + } + return rowsAffected, nil +} diff --git a/scripts/port-forward.sh b/scripts/port-forward.sh index b34a564..0f79e12 100755 --- a/scripts/port-forward.sh +++ b/scripts/port-forward.sh @@ -1,42 +1,65 @@ #!/usr/bin/env bash set -euo pipefail -NAMESPACE="chronos-infra" +INFRA_NS="chronos-infra" +APP_NS="chronos" -echo "Starting port-forwards for all infra services (background)..." -echo "Press Ctrl+C to stop all." +pids=() -# postgres → localhost:5432 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-postgres-postgresql 5432:5432 & +start_forward() { + local ns="$1" + local svc="$2" + local local_port="$3" + local remote_port="$4" -# mongodb → localhost:27017 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-mongodb 27017:27017 & + if ! kubectl get svc -n "${ns}" "${svc}" >/dev/null 2>&1; then + echo "[skip] svc/${svc} not found in namespace ${ns}" + return + fi -# redis → localhost:6379 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-redis-master 6379:6379 & + echo "[start] ${ns}/svc/${svc} ${local_port}:${remote_port}" + kubectl port-forward -n "${ns}" "svc/${svc}" "${local_port}:${remote_port}" >/tmp/chronos-port-forward-${ns}-${svc}.log 2>&1 & + pids+=("$!") +} -# kafka → localhost:9092 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-kafka 9092:9092 & +cleanup() { + if [[ ${#pids[@]} -gt 0 ]]; then + kill "${pids[@]}" 2>/dev/null || true + fi +} -# minio api → localhost:9000 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-minio 9000:9000 & +trap cleanup INT TERM EXIT -# minio ui → localhost:9001 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-minio 9001:9001 & +echo "Starting port-forwards (background). Press Ctrl+C to stop all." + +# Infra services +start_forward "${INFRA_NS}" "chronos-postgres-postgresql" 5432 5432 +start_forward "${INFRA_NS}" "chronos-mongodb" 27017 27017 +start_forward "${INFRA_NS}" "chronos-redis-master" 6379 6379 +start_forward "${INFRA_NS}" "chronos-kafka" 9092 9092 +start_forward "${INFRA_NS}" "chronos-minio" 9000 9000 +start_forward "${INFRA_NS}" "chronos-minio" 9001 9001 +start_forward "${INFRA_NS}" "chronos-kafka-ui" 8090 80 + +# App services +start_forward "${APP_NS}" "chronos-server" 8080 8080 +start_forward "${APP_NS}" "job-executor" 8081 8080 +start_forward "${APP_NS}" "bulk-ingestor" 8082 8080 +start_forward "${APP_NS}" "bulk-executor" 8083 8080 -# kafka-ui → localhost:8090 -kubectl port-forward -n "${NAMESPACE}" svc/chronos-kafka-ui 8090:80 & echo "" -echo "Forwarding:" -echo " postgres → localhost:5432" -echo " mongodb → localhost:27017" -echo " redis → localhost:6379" -echo " kafka → localhost:9092" -echo " minio → localhost:9000 (api) / localhost:9001 (ui)" -echo " kafka-ui → localhost:8090" +echo "Forwarding (if service exists):" +echo " postgres -> localhost:5432" +echo " mongodb -> localhost:27017" +echo " redis -> localhost:6379" +echo " kafka -> localhost:9092" +echo " minio -> localhost:9000 (api), localhost:9001 (console)" +echo " kafka-ui -> localhost:8090" +echo " chronos-server-> localhost:8080" +echo " job-executor -> localhost:8081" +echo " bulk-ingestor -> localhost:8082" +echo " bulk-executor -> localhost:8083" echo "" -# Wait for any background job to exit (Ctrl+C kills them all) -trap 'kill $(jobs -p) 2>/dev/null; exit 0' INT TERM wait diff --git a/scripts/smoke-test-phase7.sh b/scripts/smoke-test-phase7.sh index d89e91d..dafe8ea 100755 --- a/scripts/smoke-test-phase7.sh +++ b/scripts/smoke-test-phase7.sh @@ -100,7 +100,16 @@ if [[ -z "${FIRST_FIRED_AT}" || "${FIRST_FIRED_AT}" == "None" ]]; then fi echo "First fire observed at: ${FIRST_FIRED_AT}" -POST_FIRST_EXEC_COUNT="$(count_executions "${JOB_DEF_ID}")" +echo "Waiting for execution count to increase after first fire..." +POST_FIRST_EXEC_COUNT="${INITIAL_EXEC_COUNT}" +POST_FIRST_DEADLINE=$(( $(date -u +%s) + WAIT_FIRST_FIRE_SECONDS )) +while [[ "$(date -u +%s)" -lt "${POST_FIRST_DEADLINE}" ]]; do + POST_FIRST_EXEC_COUNT="$(count_executions "${JOB_DEF_ID}")" + if [[ "${POST_FIRST_EXEC_COUNT}" -gt "${INITIAL_EXEC_COUNT}" ]]; then + break + fi + sleep "${POLL_INTERVAL_SECONDS}" +done echo "Execution count after first fire: ${POST_FIRST_EXEC_COUNT}" if [[ "${POST_FIRST_EXEC_COUNT}" -le "${INITIAL_EXEC_COUNT}" ]]; then echo "Expected execution count to increase after first cron fire."