From 789c9613be148bd175823aa71152dca23717c6e9 Mon Sep 17 00:00:00 2001 From: "Kai (Tam Nhu) Tran" <61256810+kaitranntt@users.noreply.github.com> Date: Sun, 29 Mar 2026 01:43:19 -0400 Subject: [PATCH] fix(cron): harden toggle and scheduler state (#540) - recompute next_run_at transactionally on toggle/update instead of non-atomic read/compute/write - keep repeated EnableJob(true) idempotent by preserving existing next_run_at - reject re-enabling expired one-shot at jobs with ErrCronJobNoFutureRun - claim due jobs conditionally, reload before execution, conditional writeback - align PG and SQLite behavior with shared helpers (DRY) - atomic scheduler writeback via single UPDATE with CASE expression Closes #378 --- internal/gateway/methods/cron.go | 13 +- internal/store/cron_store.go | 208 +++++++++++ internal/store/cron_store_test.go | 113 ++++++ internal/store/pg/cron_crud.go | 42 ++- internal/store/pg/cron_scan.go | 40 +- internal/store/pg/cron_scheduler.go | 73 +++- internal/store/pg/cron_update.go | 351 +++++++++--------- internal/store/sqlitestore/cron.go | 63 +--- internal/store/sqlitestore/cron_crud.go | 370 ++++++++++--------- internal/store/sqlitestore/cron_crud_test.go | 337 +++++++++++++++++ internal/store/sqlitestore/cron_scheduler.go | 73 +++- internal/store/sqlitestore/helpers.go | 7 +- 12 files changed, 1222 insertions(+), 468 deletions(-) create mode 100644 internal/store/cron_store_test.go create mode 100644 internal/store/sqlitestore/cron_crud_test.go diff --git a/internal/gateway/methods/cron.go b/internal/gateway/methods/cron.go index 4969375e..42d8608d 100644 --- a/internal/gateway/methods/cron.go +++ b/internal/gateway/methods/cron.go @@ -3,6 +3,7 @@ package methods import ( "context" "encoding/json" + "errors" "log/slog" "regexp" @@ -155,7 +156,11 @@ func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client, } if err := m.service.EnableJob(ctx, params.JobID, params.Enabled); err != nil { - client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, err.Error())) + code := protocol.ErrInvalidRequest + if errors.Is(err, store.ErrCronJobNotFound) { + code = protocol.ErrNotFound + } + client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error())) return } @@ -200,7 +205,11 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client, job, err := m.service.UpdateJob(ctx, jobID, params.Patch) if err != nil { - client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error())) + code := protocol.ErrInvalidRequest + if errors.Is(err, store.ErrCronJobNotFound) { + code = protocol.ErrNotFound + } + client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error())) return } diff --git a/internal/store/cron_store.go b/internal/store/cron_store.go index a5e4bb24..a8ea96ff 100644 --- a/internal/store/cron_store.go +++ b/internal/store/cron_store.go @@ -2,11 +2,20 @@ package store import ( "context" + "errors" + "fmt" + "sort" "time" + "github.com/adhocore/gronx" "github.com/google/uuid" ) +var ( + ErrCronJobNotFound = errors.New("cron job not found") + ErrCronJobNoFutureRun = errors.New("cron job has no future run") +) + // CronJob represents a scheduled job. type CronJob struct { ID string `json:"id"` @@ -124,3 +133,202 @@ type CronStore interface { type CacheInvalidatable interface { InvalidateCache() } + +// CronJobMutableState holds the mutable fields of a cron job loaded within a +// transaction for read-compute-write operations (EnableJob, UpdateJob). +type CronJobMutableState struct { + Enabled bool + Schedule CronSchedule + NextRunAt *time.Time + Payload CronPayload +} + +// ComputeNextRun calculates the next run time for a cron schedule. +// defaultTZ is used for cron expressions that do not specify a per-job timezone. +func ComputeNextRun(schedule *CronSchedule, now time.Time, defaultTZ string) *time.Time { + switch schedule.Kind { + case "at": + if schedule.AtMS != nil { + t := time.UnixMilli(*schedule.AtMS) + if t.After(now) { + return &t + } + } + return nil + case "every": + if schedule.EveryMS != nil && *schedule.EveryMS > 0 { + t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond) + return &t + } + return nil + case "cron": + if schedule.Expr == "" { + return nil + } + tz := schedule.TZ + if tz == "" { + tz = defaultTZ + } + evalTime := now + if tz != "" { + if loc, err := time.LoadLocation(tz); err == nil { + evalTime = now.In(loc) + } + } + nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false) + if err != nil { + return nil + } + utcNext := nextTime.UTC() + return &utcNext + default: + return nil + } +} + +// NextRunForSchedule resolves the persisted next_run_at for a given schedule state. +func NextRunForSchedule(schedule *CronSchedule, enabled bool, now time.Time, defaultTZ string) (*time.Time, error) { + if !enabled { + return nil, nil + } + + next := ComputeNextRun(schedule, now, defaultTZ) + if next != nil { + return next, nil + } + + switch schedule.Kind { + case "at": + return nil, fmt.Errorf("%w: at schedule is already in the past", ErrCronJobNoFutureRun) + case "cron": + return nil, fmt.Errorf("%w: cron schedule has no valid next execution", ErrCronJobNoFutureRun) + case "every": + return nil, fmt.Errorf("%w: every schedule has no valid interval", ErrCronJobNoFutureRun) + default: + return nil, fmt.Errorf("%w: unsupported schedule kind %q", ErrCronJobNoFutureRun, schedule.Kind) + } +} + +// NextRunForToggle returns the next run state after explicitly enabling or +// disabling a cron job. Disabling clears next_run_at immediately so the +// scheduler stops seeing the job as runnable. +func NextRunForToggle(schedule *CronSchedule, enabled, currentlyEnabled bool, currentNextRunAt *time.Time, now time.Time, defaultTZ string) (*time.Time, error) { + if !enabled { + return nil, nil + } + if currentlyEnabled && currentNextRunAt != nil { + next := *currentNextRunAt + return &next, nil + } + return NextRunForSchedule(schedule, true, now, defaultTZ) +} + +// MergeCronSchedule applies a partial schedule patch on top of the current schedule. +func MergeCronSchedule(current CronSchedule, patch *CronSchedule) CronSchedule { + if patch == nil { + return current + } + + newKind := patch.Kind + if newKind == "" { + newKind = current.Kind + } + + merged := CronSchedule{Kind: newKind} + switch newKind { + case "cron": + if patch.Expr != "" { + merged.Expr = patch.Expr + } else if current.Kind == newKind { + merged.Expr = current.Expr + } + if patch.TZ != "" { + merged.TZ = patch.TZ + } else if current.Kind == newKind { + merged.TZ = current.TZ + } + case "every": + if patch.EveryMS != nil { + merged.EveryMS = patch.EveryMS + } else if current.Kind == newKind { + merged.EveryMS = current.EveryMS + } + case "at": + if patch.AtMS != nil { + merged.AtMS = patch.AtMS + } else if current.Kind == newKind { + merged.AtMS = current.AtMS + } + } + + return merged +} + +// ValidateCronSchedule checks structural schedule validity without evaluating future run existence. +func ValidateCronSchedule(schedule *CronSchedule) error { + switch schedule.Kind { + case "cron": + if schedule.Expr == "" { + return fmt.Errorf("cron schedule requires expr") + } + if !gronx.New().IsValid(schedule.Expr) { + return fmt.Errorf("invalid cron expression: %s", schedule.Expr) + } + if schedule.TZ != "" { + if _, err := time.LoadLocation(schedule.TZ); err != nil { + return fmt.Errorf("invalid timezone: %s", schedule.TZ) + } + } + case "every": + if schedule.EveryMS == nil || *schedule.EveryMS <= 0 { + return fmt.Errorf("every schedule requires positive everyMs") + } + case "at": + if schedule.AtMS == nil { + return fmt.Errorf("at schedule requires atMs") + } + default: + return fmt.Errorf("invalid schedule kind: %s", schedule.Kind) + } + return nil +} + +// ApplyCronScheduleUpdates populates the update map with the column values +// for a fully-resolved cron schedule (after merge + validation). +func ApplyCronScheduleUpdates(updates map[string]any, schedule CronSchedule) { + updates["schedule_kind"] = schedule.Kind + + switch schedule.Kind { + case "cron": + updates["cron_expression"] = schedule.Expr + if schedule.TZ != "" { + updates["timezone"] = schedule.TZ + } else { + updates["timezone"] = nil + } + updates["interval_ms"] = nil + updates["run_at"] = nil + case "every": + updates["cron_expression"] = nil + updates["timezone"] = nil + updates["interval_ms"] = *schedule.EveryMS + updates["run_at"] = nil + case "at": + runAt := time.UnixMilli(*schedule.AtMS) + updates["cron_expression"] = nil + updates["timezone"] = nil + updates["interval_ms"] = nil + updates["run_at"] = runAt + } +} + +// SortedUpdateColumns returns the map keys in sorted order for deterministic +// SQL generation. +func SortedUpdateColumns(updates map[string]any) []string { + cols := make([]string, 0, len(updates)) + for col := range updates { + cols = append(cols, col) + } + sort.Strings(cols) + return cols +} diff --git a/internal/store/cron_store_test.go b/internal/store/cron_store_test.go new file mode 100644 index 00000000..421d1749 --- /dev/null +++ b/internal/store/cron_store_test.go @@ -0,0 +1,113 @@ +package store + +import ( + "errors" + "testing" + "time" +) + +func TestNextRunForToggle_DisableClearsNextRun(t *testing.T) { + now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC) + schedule := &CronSchedule{ + Kind: "every", + EveryMS: int64Ptr(60_000), + } + + next, err := NextRunForToggle(schedule, false, true, timePtr(now.Add(time.Minute)), now, "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if next != nil { + t.Fatalf("expected disable toggle to clear next_run_at, got %v", next) + } +} + +func TestNextRunForToggle_EnableRecomputesEverySchedule(t *testing.T) { + now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC) + schedule := &CronSchedule{ + Kind: "every", + EveryMS: int64Ptr(60_000), + } + + next, err := NextRunForToggle(schedule, true, false, nil, now, "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if next == nil { + t.Fatal("expected enable toggle to recompute next_run_at") + } + + want := now.Add(time.Minute) + if !next.Equal(want) { + t.Fatalf("got %v, want %v", next, want) + } +} + +func TestNextRunForToggle_EnableUsesDefaultTimezoneForCronSchedule(t *testing.T) { + now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC) + schedule := &CronSchedule{ + Kind: "cron", + Expr: "0 9 * * *", + } + + next, err := NextRunForToggle(schedule, true, false, nil, now, "America/Toronto") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if next == nil { + t.Fatal("expected enable toggle to compute next_run_at for cron schedule") + } + + want := time.Date(2026, time.March, 28, 13, 0, 0, 0, time.UTC) + if !next.Equal(want) { + t.Fatalf("got %v, want %v", next, want) + } +} + +func TestNextRunForToggle_AlreadyEnabledPreservesCurrentNextRun(t *testing.T) { + now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC) + currentNextRun := now.Add(5 * time.Minute) + schedule := &CronSchedule{ + Kind: "every", + EveryMS: int64Ptr(60_000), + } + + next, err := NextRunForToggle(schedule, true, true, ¤tNextRun, now.Add(time.Minute), "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if next == nil { + t.Fatal("expected preserved next run") + } + if !next.Equal(currentNextRun) { + t.Fatalf("got %v, want %v", next, currentNextRun) + } +} + +func TestNextRunForToggle_ExpiredAtReturnsError(t *testing.T) { + now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC) + past := now.Add(-time.Minute).UnixMilli() + schedule := &CronSchedule{ + Kind: "at", + AtMS: &past, + } + + next, err := NextRunForToggle(schedule, true, false, nil, now, "") + if next != nil { + t.Fatalf("expected nil next run, got %v", next) + } + if err == nil { + t.Fatal("expected error for expired at schedule") + } + if !errors.Is(err, ErrCronJobNoFutureRun) { + t.Fatalf("got %v, want ErrCronJobNoFutureRun", err) + } +} + +func int64Ptr(v int64) *int64 { + return &v +} + +func timePtr(v time.Time) *time.Time { + return &v +} diff --git a/internal/store/pg/cron_crud.go b/internal/store/pg/cron_crud.go index db6807b7..932da59a 100644 --- a/internal/store/pg/cron_crud.go +++ b/internal/store/pg/cron_crud.go @@ -185,25 +185,35 @@ func (s *PGCronStore) EnableJob(ctx context.Context, jobID string, enabled bool) return fmt.Errorf("invalid job ID: %s", jobID) } - q := "UPDATE cron_jobs SET enabled = $1, updated_at = $2 WHERE id = $3" - args := []any{enabled, time.Now(), id} - - if !store.IsCrossTenant(ctx) { - tid := store.TenantIDFromContext(ctx) - if tid == uuid.Nil { - return fmt.Errorf("tenant_id required") - } - q += fmt.Sprintf(" AND tenant_id = $%d", len(args)+1) - args = append(args, tid) - } - - res, err := s.db.ExecContext(ctx, q, args...) + tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } - if n, _ := res.RowsAffected(); n == 0 { - return fmt.Errorf("job not found") + defer tx.Rollback() + + current, err := s.lockCronJobForMutation(ctx, tx, id, false) + if err != nil { + return err } - s.cacheLoaded = false + + now := time.Now() + nextRun, err := store.NextRunForToggle(¤t.Schedule, enabled, current.Enabled, current.NextRunAt, now, s.defaultTZ) + if err != nil { + return err + } + + if err := execCronJobUpdateTx(ctx, tx, id, map[string]any{ + "enabled": enabled, + "next_run_at": nextRun, + "updated_at": now, + }); err != nil { + return err + } + + if err := tx.Commit(); err != nil { + return err + } + + s.InvalidateCache() return nil } diff --git a/internal/store/pg/cron_scan.go b/internal/store/pg/cron_scan.go index 396369a6..a4e3906a 100644 --- a/internal/store/pg/cron_scan.go +++ b/internal/store/pg/cron_scan.go @@ -7,7 +7,6 @@ import ( "fmt" "time" - "github.com/adhocore/gronx" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -127,42 +126,5 @@ func scanCronSingleRow(row *sql.Row) (*store.CronJob, error) { // defaultTZ is the gateway-level fallback IANA timezone used when the // schedule itself does not specify a timezone (existing jobs with TZ = NULL). func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ string) *time.Time { - switch schedule.Kind { - case "at": - if schedule.AtMS != nil { - t := time.UnixMilli(*schedule.AtMS) - if t.After(now) { - return &t - } - } - return nil - case "every": - if schedule.EveryMS != nil && *schedule.EveryMS > 0 { - t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond) - return &t - } - return nil - case "cron": - if schedule.Expr == "" { - return nil - } - tz := schedule.TZ - if tz == "" { - tz = defaultTZ - } - evalTime := now - if tz != "" { - if loc, err := time.LoadLocation(tz); err == nil { - evalTime = now.In(loc) - } - } - nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false) - if err != nil { - return nil - } - utcNext := nextTime.UTC() - return &utcNext - default: - return nil - } + return store.ComputeNextRun(schedule, now, defaultTZ) } diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 995ac7d6..73452b42 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -1,6 +1,8 @@ package pg import ( + "database/sql" + "errors" "log/slog" "sync" "time" @@ -146,16 +148,20 @@ func (s *PGCronStore) checkAndRunDueJobs() { return } - // Clear next_run for all due jobs first to prevent duplicate fires + now := time.Now() + var claimedJobs []store.CronJob for _, job := range dueJobs { - if id, parseErr := uuid.Parse(job.ID); parseErr == nil { - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = NULL WHERE id = $1", id) + if id, parseErr := uuid.Parse(job.ID); parseErr == nil && s.claimDueJob(id, now) { + claimedJobs = append(claimedJobs, job) } } + if len(claimedJobs) == 0 { + return + } // Execute jobs in parallel — scheduler enforces per-session serialization var wg sync.WaitGroup - for _, job := range dueJobs { + for _, job := range claimedJobs { wg.Add(1) go func(job store.CronJob) { defer wg.Done() @@ -172,6 +178,15 @@ func (s *PGCronStore) checkAndRunDueJobs() { // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at. func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error)) { + if id, parseErr := uuid.Parse(job.ID); parseErr == nil { + freshJob, ok := s.loadClaimedJob(id) + if !ok { + slog.Info("cron job skipped after claim state changed", "id", job.ID) + return + } + job = *freshJob + } + startTime := time.Now() // Wrap handler to fit ExecuteWithRetry's (string, error) signature @@ -237,9 +252,16 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule next := computeNextRun(&schedule, now, s.defaultTZ) + var nextRunValue any + if next != nil { + nextRunValue = *next + } s.db.ExecContext(s.baseCtx, - "UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_error = $3, next_run_at = $4, updated_at = $5 WHERE id = $6", - now, status, lastError, next, now, id, + `UPDATE cron_jobs SET + last_run_at = $1, last_status = $2, last_error = $3, updated_at = $4, + next_run_at = CASE WHEN enabled = true AND next_run_at IS NULL THEN $5 ELSE next_run_at END + WHERE id = $6`, + now, status, lastError, now, nextRunValue, id, ) } @@ -251,3 +273,42 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C } s.emitEvent(evt) } + +func (s *PGCronStore) claimDueJob(id uuid.UUID, now time.Time) bool { + res, err := s.db.ExecContext( + s.baseCtx, + `UPDATE cron_jobs + SET next_run_at = NULL + WHERE id = $1 AND enabled = true AND next_run_at IS NOT NULL AND next_run_at <= $2`, + id, + now, + ) + if err != nil { + slog.Warn("cron: failed to claim due job", "id", id, "error", err) + return false + } + + n, _ := res.RowsAffected() + return n == 1 +} + +func (s *PGCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) { + row := s.db.QueryRowContext( + s.baseCtx, + `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, + created_at, updated_at + FROM cron_jobs + WHERE id = $1 AND enabled = true AND next_run_at IS NULL`, + id, + ) + job, err := scanCronSingleRow(row) + if errors.Is(err, sql.ErrNoRows) { + return nil, false + } + if err != nil { + slog.Warn("cron: failed to reload claimed job", "id", id, "error", err) + return nil, false + } + return job, true +} diff --git a/internal/store/pg/cron_update.go b/internal/store/pg/cron_update.go index 6653e7af..46ade3e1 100644 --- a/internal/store/pg/cron_update.go +++ b/internal/store/pg/cron_update.go @@ -2,11 +2,13 @@ package pg import ( "context" + "database/sql" "encoding/json" + "errors" "fmt" + "strings" "time" - "github.com/adhocore/gronx" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -18,140 +20,31 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C return nil, fmt.Errorf("invalid job ID: %s", jobID) } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + + current, err := s.lockCronJobForMutation(ctx, tx, id, true) + if err != nil { + return nil, err + } + + now := time.Now() updates := make(map[string]any) + effectiveEnabled := current.Enabled + if patch.Enabled != nil { + effectiveEnabled = *patch.Enabled + updates["enabled"] = effectiveEnabled + } + if patch.Name != "" { updates["name"] = patch.Name } - if patch.Enabled != nil { - updates["enabled"] = *patch.Enabled - } - // Build tenant WHERE suffix for internal reads (defense-in-depth). - tenantSuffix := "" - var tenantArg []any - if !store.IsCrossTenant(ctx) { - if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil { - tenantSuffix = " AND tenant_id = $2" - tenantArg = append(tenantArg, tid) - } - } - - if patch.Schedule != nil { - // Fetch current schedule to merge with patch (partial updates) - var curKind string - var curExpr, curTZ *string - var curIntervalMS *int64 - var curRunAt *time.Time - if err := s.db.QueryRowContext(ctx, - "SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = $1"+tenantSuffix, - append([]any{id}, tenantArg...)..., - ).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt); err != nil { - return nil, fmt.Errorf("failed to fetch current schedule for job %s: %w", jobID, err) - } - - // Resolve the effective schedule kind - newKind := patch.Schedule.Kind - if newKind == "" { - newKind = curKind // keep current kind if not specified - } - updates["schedule_kind"] = newKind - - // Set type-specific fields and clear others - switch newKind { - case "cron": - if patch.Schedule.Expr != "" { - updates["cron_expression"] = patch.Schedule.Expr - } - if patch.Schedule.TZ != "" { - if _, err := time.LoadLocation(patch.Schedule.TZ); err != nil { - return nil, fmt.Errorf("invalid timezone: %s", patch.Schedule.TZ) - } - updates["timezone"] = patch.Schedule.TZ - } - // Clear other type fields when switching to cron - if curKind != "cron" { - updates["interval_ms"] = nil - updates["run_at"] = nil - } - case "every": - if patch.Schedule.EveryMS != nil { - updates["interval_ms"] = *patch.Schedule.EveryMS - } - // Clear other type fields when switching to every - if curKind != "every" { - updates["cron_expression"] = nil - updates["timezone"] = nil - updates["run_at"] = nil - } - case "at": - if patch.Schedule.AtMS != nil { - t := time.UnixMilli(*patch.Schedule.AtMS) - updates["run_at"] = t - } - // Clear other type fields when switching to at - if curKind != "at" { - updates["cron_expression"] = nil - updates["timezone"] = nil - updates["interval_ms"] = nil - } - } - - // Build merged schedule for recomputing next_run_at - merged := store.CronSchedule{Kind: newKind} - switch newKind { - case "cron": - if patch.Schedule.Expr != "" { - merged.Expr = patch.Schedule.Expr - } else if curExpr != nil { - merged.Expr = *curExpr - } - if patch.Schedule.TZ != "" { - merged.TZ = patch.Schedule.TZ - } else if curTZ != nil && newKind == curKind { - merged.TZ = *curTZ - } - case "every": - if patch.Schedule.EveryMS != nil { - merged.EveryMS = patch.Schedule.EveryMS - } else if curIntervalMS != nil { - merged.EveryMS = curIntervalMS - } - case "at": - if patch.Schedule.AtMS != nil { - merged.AtMS = patch.Schedule.AtMS - } else if curRunAt != nil { - ms := curRunAt.UnixMilli() - merged.AtMS = &ms - } - } - - // Validate the merged schedule before applying - switch merged.Kind { - case "cron": - if merged.Expr == "" { - return nil, fmt.Errorf("cron schedule requires expr") - } - gx := gronx.New() - if !gx.IsValid(merged.Expr) { - return nil, fmt.Errorf("invalid cron expression: %s", merged.Expr) - } - case "every": - if merged.EveryMS == nil || *merged.EveryMS <= 0 { - return nil, fmt.Errorf("every schedule requires positive everyMs") - } - case "at": - if merged.AtMS == nil { - return nil, fmt.Errorf("at schedule requires atMs") - } - } - - next := computeNextRun(&merged, time.Now(), s.defaultTZ) - updates["next_run_at"] = next - } if patch.DeleteAfterRun != nil { updates["delete_after_run"] = *patch.DeleteAfterRun } - - // Update agent_id column if patch.AgentID != nil { if *patch.AgentID == "" { updates["agent_id"] = nil @@ -160,60 +53,170 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C } } - // Update payload JSONB — fetch current, merge patch fields, re-serialize + if patch.Schedule != nil { + merged := store.MergeCronSchedule(current.Schedule, patch.Schedule) + if err := store.ValidateCronSchedule(&merged); err != nil { + return nil, err + } + + store.ApplyCronScheduleUpdates(updates, merged) + + nextRun, err := store.NextRunForSchedule(&merged, effectiveEnabled, now, s.defaultTZ) + if err != nil { + return nil, err + } + updates["next_run_at"] = nextRun + } else if patch.Enabled != nil { + nextRun, err := store.NextRunForToggle(¤t.Schedule, effectiveEnabled, current.Enabled, current.NextRunAt, now, s.defaultTZ) + if err != nil { + return nil, err + } + updates["next_run_at"] = nextRun + } + needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil if needsPayloadUpdate { - var payloadJSON []byte - if scanErr := s.db.QueryRowContext(ctx, "SELECT payload FROM cron_jobs WHERE id = $1"+tenantSuffix, append([]any{id}, tenantArg...)...).Scan(&payloadJSON); scanErr == nil { - var payload store.CronPayload - if len(payloadJSON) > 0 { - if err := json.Unmarshal(payloadJSON, &payload); err != nil { - return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", jobID, err) - } - } - - if patch.Message != "" { - payload.Message = patch.Message - } - if patch.Deliver != nil { - payload.Deliver = *patch.Deliver - } - if patch.Channel != nil { - payload.Channel = *patch.Channel - } - if patch.To != nil { - payload.To = *patch.To - } - if patch.WakeHeartbeat != nil { - payload.WakeHeartbeat = *patch.WakeHeartbeat - } - - merged, err := json.Marshal(payload) - if err != nil { - return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err) - } - updates["payload"] = merged + payload := current.Payload + if patch.Message != "" { + payload.Message = patch.Message } + if patch.Deliver != nil { + payload.Deliver = *patch.Deliver + } + if patch.Channel != nil { + payload.Channel = *patch.Channel + } + if patch.To != nil { + payload.To = *patch.To + } + if patch.WakeHeartbeat != nil { + payload.WakeHeartbeat = *patch.WakeHeartbeat + } + + mergedPayload, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err) + } + updates["payload"] = mergedPayload } - updates["updated_at"] = time.Now() - - var execErr error - if store.IsCrossTenant(ctx) { - execErr = execMapUpdate(ctx, s.db, "cron_jobs", id, updates) - } else { - tid := store.TenantIDFromContext(ctx) - if tid == uuid.Nil { - execErr = fmt.Errorf("tenant_id required for update") - } else { - execErr = execMapUpdateWhereTenant(ctx, s.db, "cron_jobs", updates, id, tid) - } - } - if err := execErr; err != nil { + updates["updated_at"] = now + if err := execCronJobUpdateTx(ctx, tx, id, updates); err != nil { return nil, err } - s.cacheLoaded = false + if err := tx.Commit(); err != nil { + return nil, err + } + + s.InvalidateCache() job, _ := s.scanJob(ctx, id) return job, nil } + +func (s *PGCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id uuid.UUID, loadPayload bool) (*store.CronJobMutableState, error) { + q := `SELECT enabled, schedule_kind, cron_expression, run_at, timezone, interval_ms, next_run_at, payload + FROM cron_jobs WHERE id = $1` + args := []any{id} + + if !store.IsCrossTenant(ctx) { + tid := store.TenantIDFromContext(ctx) + if tid == uuid.Nil { + return nil, fmt.Errorf("tenant_id required") + } + q += fmt.Sprintf(" AND tenant_id = $%d", len(args)+1) + args = append(args, tid) + } + q += " FOR UPDATE" + + var ( + state store.CronJobMutableState + scheduleKind string + cronExpr *string + runAt *time.Time + tz *string + intervalMS *int64 + nextRunAt *time.Time + payloadJSON []byte + ) + + if err := tx.QueryRowContext(ctx, q, args...).Scan( + &state.Enabled, + &scheduleKind, + &cronExpr, + &runAt, + &tz, + &intervalMS, + &nextRunAt, + &payloadJSON, + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, store.ErrCronJobNotFound + } + return nil, err + } + + state.Schedule = store.CronSchedule{Kind: scheduleKind} + if cronExpr != nil { + state.Schedule.Expr = *cronExpr + } + if runAt != nil { + ms := runAt.UnixMilli() + state.Schedule.AtMS = &ms + } + if tz != nil { + state.Schedule.TZ = *tz + } + if intervalMS != nil { + state.Schedule.EveryMS = intervalMS + } + state.NextRunAt = nextRunAt + + if loadPayload && len(payloadJSON) > 0 { + if err := json.Unmarshal(payloadJSON, &state.Payload); err != nil { + return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", id, err) + } + } + + return &state, nil +} + + +func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error { + if len(updates) == 0 { + return nil + } + + var ( + setClauses []string + args []any + ) + for idx, col := range store.SortedUpdateColumns(updates) { + if !validColumnName.MatchString(col) { + return fmt.Errorf("invalid column name: %q", col) + } + setClauses = append(setClauses, fmt.Sprintf("%s = $%d", col, idx+1)) + args = append(args, updates[col]) + } + + args = append(args, id) + q := fmt.Sprintf("UPDATE cron_jobs SET %s WHERE id = $%d", strings.Join(setClauses, ", "), len(args)) + if !store.IsCrossTenant(ctx) { + tid := store.TenantIDFromContext(ctx) + if tid == uuid.Nil { + return fmt.Errorf("tenant_id required") + } + args = append(args, tid) + q += fmt.Sprintf(" AND tenant_id = $%d", len(args)) + } + + res, err := tx.ExecContext(ctx, q, args...) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return store.ErrCronJobNotFound + } + return nil +} + diff --git a/internal/store/sqlitestore/cron.go b/internal/store/sqlitestore/cron.go index 3f959b4f..5a47031b 100644 --- a/internal/store/sqlitestore/cron.go +++ b/internal/store/sqlitestore/cron.go @@ -11,7 +11,6 @@ import ( "sync" "time" - "github.com/adhocore/gronx" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/cron" @@ -24,6 +23,7 @@ const defaultCronCacheTTL = 2 * time.Minute type SQLiteCronStore struct { db *sql.DB mu sync.Mutex + writeMu sync.Mutex baseCtx context.Context cancelCtx context.CancelFunc onJob func(job *store.CronJob) (*store.CronJobResult, error) @@ -119,14 +119,14 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { var name, scheduleKind string var enabled, deleteAfterRun bool var cronExpr, tz, lastStatus, lastError *string - var runAt, nextRunAt, lastRunAt *time.Time + var runAt, nextRunAt, lastRunAt nullSqliteTime var intervalMS *int64 var payloadJSON []byte - var createdAt, updatedAt time.Time + createdAt, updatedAt := scanTimePair() err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz, &intervalMS, &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError, - &createdAt, &updatedAt) + createdAt, updatedAt) if err != nil { return nil, err } @@ -145,8 +145,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { Enabled: enabled, Schedule: store.CronSchedule{Kind: scheduleKind}, Payload: payload, - CreatedAtMS: createdAt.UnixMilli(), - UpdatedAtMS: updatedAt.UnixMilli(), + CreatedAtMS: createdAt.Time.UnixMilli(), + UpdatedAtMS: updatedAt.Time.UnixMilli(), DeleteAfterRun: deleteAfterRun, } @@ -159,8 +159,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { if cronExpr != nil { job.Schedule.Expr = *cronExpr } - if runAt != nil { - ms := runAt.UnixMilli() + if runAt.Valid { + ms := runAt.Time.UnixMilli() job.Schedule.AtMS = &ms } if intervalMS != nil { @@ -169,12 +169,12 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { if tz != nil { job.Schedule.TZ = *tz } - if nextRunAt != nil { - ms := nextRunAt.UnixMilli() + if nextRunAt.Valid { + ms := nextRunAt.Time.UnixMilli() job.State.NextRunAtMS = &ms } - if lastRunAt != nil { - ms := lastRunAt.UnixMilli() + if lastRunAt.Valid { + ms := lastRunAt.Time.UnixMilli() job.State.LastRunAtMS = &ms } if lastStatus != nil { @@ -189,44 +189,7 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) { // computeNextRun calculates the next run time for a schedule. func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ string) *time.Time { - switch schedule.Kind { - case "at": - if schedule.AtMS != nil { - t := time.UnixMilli(*schedule.AtMS) - if t.After(now) { - return &t - } - } - return nil - case "every": - if schedule.EveryMS != nil && *schedule.EveryMS > 0 { - t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond) - return &t - } - return nil - case "cron": - if schedule.Expr == "" { - return nil - } - tz := schedule.TZ - if tz == "" { - tz = defaultTZ - } - evalTime := now - if tz != "" { - if loc, err := time.LoadLocation(tz); err == nil { - evalTime = now.In(loc) - } - } - nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false) - if err != nil { - return nil - } - utcNext := nextTime.UTC() - return &utcNext - default: - return nil - } + return store.ComputeNextRun(schedule, now, defaultTZ) } func (s *SQLiteCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob, error) { diff --git a/internal/store/sqlitestore/cron_crud.go b/internal/store/sqlitestore/cron_crud.go index 3a3c3023..0dfd86ab 100644 --- a/internal/store/sqlitestore/cron_crud.go +++ b/internal/store/sqlitestore/cron_crud.go @@ -4,12 +4,14 @@ package sqlitestore import ( "context" + "database/sql" "encoding/json" + "errors" "fmt" "log/slog" + "strings" "time" - "github.com/adhocore/gronx" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -184,25 +186,38 @@ func (s *SQLiteCronStore) EnableJob(ctx context.Context, jobID string, enabled b return fmt.Errorf("invalid job ID: %s", jobID) } - q := "UPDATE cron_jobs SET enabled = ?, updated_at = ? WHERE id = ?" - args := []any{enabled, time.Now(), id} + s.writeMu.Lock() + defer s.writeMu.Unlock() - if !store.IsCrossTenant(ctx) { - tid := store.TenantIDFromContext(ctx) - if tid == uuid.Nil { - return fmt.Errorf("tenant_id required") - } - q += " AND tenant_id = ?" - args = append(args, tid) - } - - res, err := s.db.ExecContext(ctx, q, args...) + tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } - if n, _ := res.RowsAffected(); n == 0 { - return fmt.Errorf("job not found") + defer tx.Rollback() + + current, err := s.lockCronJobForMutation(ctx, tx, id, false) + if err != nil { + return err } + + now := time.Now() + nextRun, err := store.NextRunForToggle(¤t.Schedule, enabled, current.Enabled, current.NextRunAt, now, s.defaultTZ) + if err != nil { + return err + } + + if err := execCronJobUpdateTx(ctx, tx, id, map[string]any{ + "enabled": enabled, + "next_run_at": nextRun, + "updated_at": now, + }); err != nil { + return err + } + + if err := tx.Commit(); err != nil { + return err + } + s.InvalidateCache() return nil } @@ -213,86 +228,31 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto return nil, fmt.Errorf("invalid job ID: %s", jobID) } + s.writeMu.Lock() + defer s.writeMu.Unlock() + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + + current, err := s.lockCronJobForMutation(ctx, tx, id, true) + if err != nil { + return nil, err + } + + now := time.Now() updates := make(map[string]any) + effectiveEnabled := current.Enabled + if patch.Enabled != nil { + effectiveEnabled = *patch.Enabled + updates["enabled"] = effectiveEnabled + } + if patch.Name != "" { updates["name"] = patch.Name } - if patch.Enabled != nil { - updates["enabled"] = *patch.Enabled - } - - // Build tenant WHERE suffix for schedule reads. - tenantSuffix := "" - var tenantArg []any - if !store.IsCrossTenant(ctx) { - if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil { - tenantSuffix = " AND tenant_id = ?" - tenantArg = append(tenantArg, tid) - } - } - - if patch.Schedule != nil { - var curKind string - var curExpr, curTZ *string - var curIntervalMS *int64 - var curRunAt *time.Time - if err := s.db.QueryRowContext(ctx, - "SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = ?"+tenantSuffix, - append([]any{id}, tenantArg...)..., - ).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt); err != nil { - return nil, fmt.Errorf("failed to fetch current schedule for job %s: %w", jobID, err) - } - - newKind := patch.Schedule.Kind - if newKind == "" { - newKind = curKind - } - updates["schedule_kind"] = newKind - - switch newKind { - case "cron": - if patch.Schedule.Expr != "" { - updates["cron_expression"] = patch.Schedule.Expr - } - if patch.Schedule.TZ != "" { - if _, err := time.LoadLocation(patch.Schedule.TZ); err != nil { - return nil, fmt.Errorf("invalid timezone: %s", patch.Schedule.TZ) - } - updates["timezone"] = patch.Schedule.TZ - } - if curKind != "cron" { - updates["interval_ms"] = nil - updates["run_at"] = nil - } - case "every": - if patch.Schedule.EveryMS != nil { - updates["interval_ms"] = *patch.Schedule.EveryMS - } - if curKind != "every" { - updates["cron_expression"] = nil - updates["timezone"] = nil - updates["run_at"] = nil - } - case "at": - if patch.Schedule.AtMS != nil { - t := time.UnixMilli(*patch.Schedule.AtMS) - updates["run_at"] = t - } - if curKind != "at" { - updates["cron_expression"] = nil - updates["timezone"] = nil - updates["interval_ms"] = nil - } - } - - merged := buildMergedSchedule(newKind, curKind, patch.Schedule, curExpr, curTZ, curIntervalMS, curRunAt) - if err := validateSchedule(&merged); err != nil { - return nil, err - } - next := computeNextRun(&merged, time.Now(), s.defaultTZ) - updates["next_run_at"] = next - } - if patch.DeleteAfterRun != nil { updates["delete_after_run"] = *patch.DeleteAfterRun } @@ -304,54 +264,60 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto } } + if patch.Schedule != nil { + merged := store.MergeCronSchedule(current.Schedule, patch.Schedule) + if err := store.ValidateCronSchedule(&merged); err != nil { + return nil, err + } + + store.ApplyCronScheduleUpdates(updates, merged) + + nextRun, err := store.NextRunForSchedule(&merged, effectiveEnabled, now, s.defaultTZ) + if err != nil { + return nil, err + } + updates["next_run_at"] = nextRun + } else if patch.Enabled != nil { + nextRun, err := store.NextRunForToggle(¤t.Schedule, effectiveEnabled, current.Enabled, current.NextRunAt, now, s.defaultTZ) + if err != nil { + return nil, err + } + updates["next_run_at"] = nextRun + } + needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil if needsPayloadUpdate { - var payloadJSON []byte - if scanErr := s.db.QueryRowContext(ctx, "SELECT payload FROM cron_jobs WHERE id = ?"+tenantSuffix, append([]any{id}, tenantArg...)...).Scan(&payloadJSON); scanErr == nil { - var p store.CronPayload - if len(payloadJSON) > 0 { - if err := json.Unmarshal(payloadJSON, &p); err != nil { - return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", jobID, err) - } - } - if patch.Message != "" { - p.Message = patch.Message - } - if patch.Deliver != nil { - p.Deliver = *patch.Deliver - } - if patch.Channel != nil { - p.Channel = *patch.Channel - } - if patch.To != nil { - p.To = *patch.To - } - if patch.WakeHeartbeat != nil { - p.WakeHeartbeat = *patch.WakeHeartbeat - } - merged, err := json.Marshal(p) - if err != nil { - return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err) - } - updates["payload"] = merged + payload := current.Payload + if patch.Message != "" { + payload.Message = patch.Message } + if patch.Deliver != nil { + payload.Deliver = *patch.Deliver + } + if patch.Channel != nil { + payload.Channel = *patch.Channel + } + if patch.To != nil { + payload.To = *patch.To + } + if patch.WakeHeartbeat != nil { + payload.WakeHeartbeat = *patch.WakeHeartbeat + } + + mergedPayload, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err) + } + updates["payload"] = mergedPayload } - updates["updated_at"] = time.Now() - - var execErr error - if store.IsCrossTenant(ctx) { - execErr = execMapUpdate(ctx, s.db, "cron_jobs", id, updates) - } else { - tid := store.TenantIDFromContext(ctx) - if tid == uuid.Nil { - execErr = fmt.Errorf("tenant_id required for update") - } else { - execErr = execMapUpdateWhereTenant(ctx, s.db, "cron_jobs", updates, id, tid) - } + updates["updated_at"] = now + if err := execCronJobUpdateTx(ctx, tx, id, updates); err != nil { + return nil, err } - if execErr != nil { - return nil, execErr + + if err := tx.Commit(); err != nil { + return nil, err } s.InvalidateCache() @@ -359,55 +325,111 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto return job, nil } -func buildMergedSchedule(newKind, curKind string, patch *store.CronSchedule, curExpr, curTZ *string, curIntervalMS *int64, curRunAt *time.Time) store.CronSchedule { - merged := store.CronSchedule{Kind: newKind} - switch newKind { - case "cron": - if patch.Expr != "" { - merged.Expr = patch.Expr - } else if curExpr != nil { - merged.Expr = *curExpr +func (s *SQLiteCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id uuid.UUID, loadPayload bool) (*store.CronJobMutableState, error) { + q := `SELECT enabled, schedule_kind, cron_expression, run_at, timezone, interval_ms, next_run_at, payload + FROM cron_jobs WHERE id = ?` + args := []any{id} + + if !store.IsCrossTenant(ctx) { + tid := store.TenantIDFromContext(ctx) + if tid == uuid.Nil { + return nil, fmt.Errorf("tenant_id required") } - if patch.TZ != "" { - merged.TZ = patch.TZ - } else if curTZ != nil && newKind == curKind { - merged.TZ = *curTZ + q += " AND tenant_id = ?" + args = append(args, tid) + } + + var ( + state store.CronJobMutableState + scheduleKind string + cronExpr *string + runAt nullSqliteTime + tz *string + intervalMS *int64 + nextRunAt nullSqliteTime + payloadJSON []byte + ) + + if err := tx.QueryRowContext(ctx, q, args...).Scan( + &state.Enabled, + &scheduleKind, + &cronExpr, + &runAt, + &tz, + &intervalMS, + &nextRunAt, + &payloadJSON, + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, store.ErrCronJobNotFound } - case "every": - if patch.EveryMS != nil { - merged.EveryMS = patch.EveryMS - } else if curIntervalMS != nil { - merged.EveryMS = curIntervalMS - } - case "at": - if patch.AtMS != nil { - merged.AtMS = patch.AtMS - } else if curRunAt != nil { - ms := curRunAt.UnixMilli() - merged.AtMS = &ms + return nil, err + } + + state.Schedule = store.CronSchedule{Kind: scheduleKind} + if cronExpr != nil { + state.Schedule.Expr = *cronExpr + } + if runAt.Valid { + ms := runAt.Time.UnixMilli() + state.Schedule.AtMS = &ms + } + if tz != nil { + state.Schedule.TZ = *tz + } + if intervalMS != nil { + state.Schedule.EveryMS = intervalMS + } + if nextRunAt.Valid { + next := nextRunAt.Time + state.NextRunAt = &next + } + + if loadPayload && len(payloadJSON) > 0 { + if err := json.Unmarshal(payloadJSON, &state.Payload); err != nil { + return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", id, err) } } - return merged + + return &state, nil } -func validateSchedule(s *store.CronSchedule) error { - switch s.Kind { - case "cron": - if s.Expr == "" { - return fmt.Errorf("cron schedule requires expr") + +func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error { + if len(updates) == 0 { + return nil + } + + var ( + setClauses []string + args []any + ) + for _, col := range store.SortedUpdateColumns(updates) { + if !validColumnName.MatchString(col) { + return fmt.Errorf("invalid column name: %q", col) } - gx := gronx.New() - if !gx.IsValid(s.Expr) { - return fmt.Errorf("invalid cron expression: %s", s.Expr) - } - case "every": - if s.EveryMS == nil || *s.EveryMS <= 0 { - return fmt.Errorf("every schedule requires positive everyMs") - } - case "at": - if s.AtMS == nil { - return fmt.Errorf("at schedule requires atMs") + setClauses = append(setClauses, col+" = ?") + args = append(args, sqliteVal(updates[col])) + } + + args = append(args, id) + q := "UPDATE cron_jobs SET " + strings.Join(setClauses, ", ") + " WHERE id = ?" + if !store.IsCrossTenant(ctx) { + tid := store.TenantIDFromContext(ctx) + if tid == uuid.Nil { + return fmt.Errorf("tenant_id required") } + args = append(args, tid) + q += " AND tenant_id = ?" + } + + res, err := tx.ExecContext(ctx, q, args...) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return store.ErrCronJobNotFound } return nil } + diff --git a/internal/store/sqlitestore/cron_crud_test.go b/internal/store/sqlitestore/cron_crud_test.go new file mode 100644 index 00000000..c9b1c8b9 --- /dev/null +++ b/internal/store/sqlitestore/cron_crud_test.go @@ -0,0 +1,337 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestSQLiteCronStore_ReenableRestoresNextRun(t *testing.T) { + cronStore, ctx, _ := newTestSQLiteCronStore(t) + everyMS := int64(time.Minute / time.Millisecond) + + job, err := cronStore.AddJob(ctx, "job-reenable", store.CronSchedule{ + Kind: "every", + EveryMS: &everyMS, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + + if err := cronStore.EnableJob(ctx, job.ID, false); err != nil { + t.Fatalf("disable error: %v", err) + } + disabled, ok := cronStore.GetJob(ctx, job.ID) + if !ok { + t.Fatal("disabled job not found") + } + if disabled.Enabled { + t.Fatal("expected job to be disabled") + } + if disabled.State.NextRunAtMS != nil { + t.Fatalf("expected disabled job to clear next_run_at, got %v", *disabled.State.NextRunAtMS) + } + + if err := cronStore.EnableJob(ctx, job.ID, true); err != nil { + t.Fatalf("re-enable error: %v", err) + } + reenabled, ok := cronStore.GetJob(ctx, job.ID) + if !ok { + t.Fatal("re-enabled job not found") + } + if !reenabled.Enabled { + t.Fatal("expected job to be enabled") + } + if reenabled.State.NextRunAtMS == nil { + t.Fatal("expected re-enabled job to have next_run_at") + } + + due := cronStore.GetDueJobs(time.UnixMilli(*reenabled.State.NextRunAtMS)) + if len(due) != 1 || due[0].ID != job.ID { + t.Fatalf("expected job %s to be due after re-enable, got %#v", job.ID, due) + } +} + +func TestSQLiteCronStore_EnableAlreadyEnabledPreservesNextRun(t *testing.T) { + cronStore, ctx, _ := newTestSQLiteCronStore(t) + everyMS := int64(time.Hour / time.Millisecond) + + job, err := cronStore.AddJob(ctx, "job-idempotent", store.CronSchedule{ + Kind: "every", + EveryMS: &everyMS, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + if job.State.NextRunAtMS == nil { + t.Fatal("expected new job to have next_run_at") + } + + originalNextRun := *job.State.NextRunAtMS + time.Sleep(20 * time.Millisecond) + + if err := cronStore.EnableJob(ctx, job.ID, true); err != nil { + t.Fatalf("enable error: %v", err) + } + + current, ok := cronStore.GetJob(ctx, job.ID) + if !ok { + t.Fatal("job not found after enable") + } + if current.State.NextRunAtMS == nil { + t.Fatal("expected enabled job to keep next_run_at") + } + if got := *current.State.NextRunAtMS; got != originalNextRun { + t.Fatalf("got next_run_at %d, want preserved %d", got, originalNextRun) + } +} + +func TestSQLiteCronStore_EnableExpiredAtReturnsError(t *testing.T) { + cronStore, ctx, _ := newTestSQLiteCronStore(t) + future := time.Now().Add(time.Hour).UnixMilli() + + job, err := cronStore.AddJob(ctx, "job-expired-at", store.CronSchedule{ + Kind: "at", + AtMS: &future, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + + if err := cronStore.EnableJob(ctx, job.ID, false); err != nil { + t.Fatalf("disable error: %v", err) + } + + past := time.Now().Add(-time.Hour).UnixMilli() + if _, err := cronStore.UpdateJob(ctx, job.ID, store.CronJobPatch{ + Schedule: &store.CronSchedule{Kind: "at", AtMS: &past}, + }); err != nil { + t.Fatalf("UpdateJob error: %v", err) + } + + err = cronStore.EnableJob(ctx, job.ID, true) + if !errors.Is(err, store.ErrCronJobNoFutureRun) { + t.Fatalf("got %v, want ErrCronJobNoFutureRun", err) + } + + current, ok := cronStore.GetJob(ctx, job.ID) + if !ok { + t.Fatal("job not found after failed enable") + } + if current.Enabled { + t.Fatal("expected failed enable to leave job disabled") + } + if current.State.NextRunAtMS != nil { + t.Fatalf("expected failed enable to leave next_run_at nil, got %v", *current.State.NextRunAtMS) + } +} + +func TestSQLiteCronStore_SchedulerSkipsDisabledCachedJob(t *testing.T) { + cronStore, ctx, db := newTestSQLiteCronStore(t) + everyMS := int64(time.Minute / time.Millisecond) + + job, err := cronStore.AddJob(ctx, "job-stale-cache", store.CronSchedule{ + Kind: "every", + EveryMS: &everyMS, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + + jobUUID := uuid.MustParse(job.ID) + pastRun := time.Now().Add(-time.Minute) + if _, err := db.ExecContext(ctx, + "UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ? AND tenant_id = ?", + pastRun, time.Now(), jobUUID, store.MasterTenantID, + ); err != nil { + t.Fatalf("mark due error: %v", err) + } + + cached, ok := cronStore.GetJob(ctx, job.ID) + if !ok { + t.Fatal("cached job not found") + } + pastMS := pastRun.UnixMilli() + cached.Enabled = true + cached.State.NextRunAtMS = &pastMS + cronStore.jobCache = []store.CronJob{*cached} + cronStore.cacheLoaded = true + cronStore.cacheTime = time.Now() + cronStore.cacheTTL = time.Hour + cronStore.baseCtx = context.Background() + + var runs atomic.Int32 + cronStore.onJob = func(job *store.CronJob) (*store.CronJobResult, error) { + runs.Add(1) + return &store.CronJobResult{Content: "ok"}, nil + } + + if _, err := db.ExecContext(ctx, + "UPDATE cron_jobs SET enabled = 0, next_run_at = NULL, updated_at = ? WHERE id = ? AND tenant_id = ?", + time.Now(), jobUUID, store.MasterTenantID, + ); err != nil { + t.Fatalf("disable error: %v", err) + } + + cronStore.checkAndRunDueJobs() + + if got := runs.Load(); got != 0 { + t.Fatalf("expected claimed disabled job to be skipped, got %d executions", got) + } +} + +func TestSQLiteCronStore_ExecuteOneJob_DoesNotRestoreNextRunAfterDisable(t *testing.T) { + cronStore, ctx, db := newTestSQLiteCronStore(t) + everyMS := int64(time.Minute / time.Millisecond) + + job, err := cronStore.AddJob(ctx, "job-midrun-disable", store.CronSchedule{ + Kind: "every", + EveryMS: &everyMS, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + + jobUUID := uuid.MustParse(job.ID) + if _, err := db.ExecContext(ctx, + "UPDATE cron_jobs SET next_run_at = NULL, updated_at = ? WHERE id = ? AND tenant_id = ?", + time.Now(), jobUUID, store.MasterTenantID, + ); err != nil { + t.Fatalf("claim setup error: %v", err) + } + + cronStore.executeOneJob(*job, func(job *store.CronJob) (*store.CronJobResult, error) { + if err := cronStore.EnableJob(ctx, job.ID, false); err != nil { + t.Fatalf("disable during run error: %v", err) + } + return &store.CronJobResult{Content: "ok"}, nil + }) + + current := mustRawJob(t, db, jobUUID) + if current.enabled { + t.Fatal("expected job to remain disabled after run") + } + if current.nextRunAt != nil { + t.Fatalf("expected disabled job to keep next_run_at nil, got %v", current.nextRunAt) + } +} + +func TestSQLiteCronStore_EnableJob_IgnoresMalformedPayload(t *testing.T) { + cronStore, ctx, db := newTestSQLiteCronStore(t) + everyMS := int64(time.Minute / time.Millisecond) + + job, err := cronStore.AddJob(ctx, "job-bad-payload", store.CronSchedule{ + Kind: "every", + EveryMS: &everyMS, + }, "hello", false, "", "", "", "user-1") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + if job == nil { + job = mustOnlyJob(t, cronStore, ctx) + } + + jobUUID := uuid.MustParse(job.ID) + if _, err := db.ExecContext(ctx, + "UPDATE cron_jobs SET payload = ? WHERE id = ? AND tenant_id = ?", + "{", jobUUID, store.MasterTenantID, + ); err != nil { + t.Fatalf("corrupt payload error: %v", err) + } + + if err := cronStore.EnableJob(ctx, job.ID, false); err != nil { + t.Fatalf("toggle with malformed payload error: %v", err) + } + + current := mustRawJob(t, db, jobUUID) + if current.enabled { + t.Fatal("expected malformed-payload job to be disabled") + } + if current.nextRunAt != nil { + t.Fatalf("expected malformed-payload job to clear next_run_at, got %v", current.nextRunAt) + } +} + +func newTestSQLiteCronStore(t *testing.T) (*SQLiteCronStore, context.Context, *sql.DB) { + t.Helper() + + db, err := OpenDB(filepath.Join(t.TempDir(), "cron.db")) + if err != nil { + t.Fatalf("OpenDB error: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema error: %v", err) + } + + ctx := store.WithCrossTenant(store.WithTenantID(context.Background(), store.MasterTenantID)) + cronStore := NewSQLiteCronStore(db) + cronStore.baseCtx = context.Background() + cronStore.cacheTTL = time.Hour + return cronStore, ctx, db +} + +func mustOnlyJob(t *testing.T, cronStore *SQLiteCronStore, ctx context.Context) *store.CronJob { + t.Helper() + + jobs := cronStore.ListJobs(ctx, true, "", "") + if len(jobs) != 1 { + t.Fatalf("expected exactly 1 job, got %#v", jobs) + } + + job := jobs[0] + return &job +} + +type rawCronJobState struct { + enabled bool + nextRunAt *time.Time +} + +func mustRawJob(t *testing.T, db *sql.DB, id uuid.UUID) rawCronJobState { + t.Helper() + + var ( + enabled bool + nextRunAt nullSqliteTime + ) + if err := db.QueryRow( + "SELECT enabled, next_run_at FROM cron_jobs WHERE id = ?", + id, + ).Scan(&enabled, &nextRunAt); err != nil { + t.Fatalf("raw cron job query error: %v", err) + } + + state := rawCronJobState{enabled: enabled} + if nextRunAt.Valid { + next := nextRunAt.Time + state.nextRunAt = &next + } + return state +} diff --git a/internal/store/sqlitestore/cron_scheduler.go b/internal/store/sqlitestore/cron_scheduler.go index a3bd32eb..b62f9e0e 100644 --- a/internal/store/sqlitestore/cron_scheduler.go +++ b/internal/store/sqlitestore/cron_scheduler.go @@ -3,6 +3,8 @@ package sqlitestore import ( + "database/sql" + "errors" "log/slog" "sync" "time" @@ -152,16 +154,20 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() { return } - // Clear next_run for all due jobs first to prevent duplicate fires. + now := time.Now() + var claimedJobs []store.CronJob for _, job := range dueJobs { - if id, parseErr := uuid.Parse(job.ID); parseErr == nil { - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = NULL WHERE id = ?", id) + if id, parseErr := uuid.Parse(job.ID); parseErr == nil && s.claimDueJob(id, now) { + claimedJobs = append(claimedJobs, job) } } + if len(claimedJobs) == 0 { + return + } // Execute jobs in parallel — scheduler enforces per-session serialization. var wg sync.WaitGroup - for _, job := range dueJobs { + for _, job := range claimedJobs { wg.Add(1) go func(job store.CronJob) { defer wg.Done() @@ -177,6 +183,15 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() { // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at. func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error)) { + if id, parseErr := uuid.Parse(job.ID); parseErr == nil { + freshJob, ok := s.loadClaimedJob(id) + if !ok { + slog.Info("cron job skipped after claim state changed", "id", job.ID) + return + } + job = *freshJob + } + startTime := time.Now() var lastResult *store.CronJobResult @@ -238,9 +253,16 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule next := computeNextRun(&schedule, now, s.defaultTZ) + var nextRunValue any + if next != nil { + nextRunValue = *next + } s.db.ExecContext(s.baseCtx, - "UPDATE cron_jobs SET last_run_at = ?, last_status = ?, last_error = ?, next_run_at = ?, updated_at = ? WHERE id = ?", - now, status, lastError, next, now, id, + `UPDATE cron_jobs SET + last_run_at = ?, last_status = ?, last_error = ?, updated_at = ?, + next_run_at = CASE WHEN enabled = 1 AND next_run_at IS NULL THEN ? ELSE next_run_at END + WHERE id = ?`, + now, status, lastError, now, nextRunValue, id, ) } @@ -251,3 +273,42 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto } s.emitEvent(evt) } + +func (s *SQLiteCronStore) claimDueJob(id uuid.UUID, now time.Time) bool { + res, err := s.db.ExecContext( + s.baseCtx, + `UPDATE cron_jobs + SET next_run_at = NULL + WHERE id = ? AND enabled = 1 AND next_run_at IS NOT NULL AND next_run_at <= ?`, + id, + now, + ) + if err != nil { + slog.Warn("cron: failed to claim due job", "id", id, "error", err) + return false + } + + n, _ := res.RowsAffected() + return n == 1 +} + +func (s *SQLiteCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) { + row := s.db.QueryRowContext( + s.baseCtx, + `SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone, + interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error, + created_at, updated_at + FROM cron_jobs + WHERE id = ? AND enabled = 1 AND next_run_at IS NULL`, + id, + ) + job, err := scanCronRow(row) + if errors.Is(err, sql.ErrNoRows) { + return nil, false + } + if err != nil { + slog.Warn("cron: failed to reload claimed job", "id", id, "error", err) + return nil, false + } + return job, true +} diff --git a/internal/store/sqlitestore/helpers.go b/internal/store/sqlitestore/helpers.go index e8d70b10..0605add8 100644 --- a/internal/store/sqlitestore/helpers.go +++ b/internal/store/sqlitestore/helpers.go @@ -119,9 +119,14 @@ func sqliteVal(v any) any { if v == nil { return nil } - switch v.(type) { + switch typed := v.(type) { case string, int, int64, float64, bool, time.Time, []byte, json.RawMessage: return v + case *time.Time: + if typed == nil { + return nil + } + return *typed } // For maps, slices, etc. — marshal to JSON string. b, err := json.Marshal(v)