mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-14 13:49:18 +02:00
* fix(scheduler): advance autopilot next_run_at after each scheduled dispatch The display-only autopilot_trigger.next_run_at column was written only on trigger create/update and never advanced afterward, so for a recurring schedule it froze at a past slot and the list rendered it as a 'next run' in the past (e.g. '53m ago'). The intended AdvanceTriggerNextRun query was dead code with zero callers. Wire it up at the scheduler's existing post-dispatch seam (replacing the last_fired_at-only TouchAutopilotTriggerFiredAt bump, which AdvanceTrigger- NextRun already supersets). The advanced value is computed on the app local clock via ComputeNextRun — the same path create/update use — so the whole next_run_at display column is owned by one clock and stays consistent; scheduling itself is untouched and still runs off DB time via NextOccurrencesUTC. On a cron/timezone parse failure we fall back to the last_fired_at-only bump. Adds a deterministic regression test for the reported scenario (hourly cron in America/New_York) and documents the local-clock ownership on ComputeNextRun. MUL-3749 Co-authored-by: multica-agent <github@multica.ai> * fix(scheduler): floor next_run_at advance at plan_time to survive clock skew Addresses review feedback on the next_run_at write-back (MUL-3749): - The post-dispatch advance computed the value from time.Now() alone. The handler is entered only after DB time judged the plan due, so if this app instance's clock lags the DB clock at a period boundary, time.Now() could recompute the slot that just fired and next_run_at would not advance — the original staleness bug, at the boundary. Extract advancedNextRun, which anchors at max(now, plan_time) via NextOccurrenceAfterUTC so the written value is always strictly after the fired plan_time while still tracking the local clock in the normal case. - Add scheduler-layer tests asserting the written value is strictly after plan_time across skew / on-slot / normal cases. The previous service-layer test only exercised the helper with an explicit after, not this path. - Sync the stale ListSchedulableAutopilotTriggers comment: the scheduler now writes last_fired_at via AdvanceTriggerNextRun (sqlc regenerated). MUL-3749 Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: J <j@multica.ai> Co-authored-by: multica-agent <github@multica.ai>
103 lines
3.8 KiB
Go
103 lines
3.8 KiB
Go
package service
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/robfig/cron/v3"
|
|
)
|
|
|
|
// cronParser accepts standard 5-field cron expressions.
|
|
var cronParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
|
|
|
|
// NextOccurrenceAfterUTC parses cronExpr in the named IANA timezone and
|
|
// returns the next activation strictly after `after`. The result is always
|
|
// in UTC and represents the canonical fire time of the next occurrence.
|
|
//
|
|
// `after` is interpreted as an absolute instant; callers should pass DB
|
|
// time (e.g. `SELECT now()`) rather than `time.Now()` so that two
|
|
// app instances with skewed clocks still produce the same answer.
|
|
//
|
|
// This is the building block the new scheduler.AutopilotScheduleDispatchJob
|
|
// uses to compute plan_times; the handler / UI write paths use it via
|
|
// ComputeNextRunFromUTC to fill in the display-only
|
|
// autopilot_trigger.next_run_at column.
|
|
func NextOccurrenceAfterUTC(cronExpr, timezone string, after time.Time) (time.Time, error) {
|
|
sched, loc, err := parseCronSchedule(cronExpr, timezone)
|
|
if err != nil {
|
|
return time.Time{}, err
|
|
}
|
|
return sched.Next(after.In(loc)).UTC(), nil
|
|
}
|
|
|
|
// NextOccurrencesUTC parses cronExpr in `timezone` and returns every
|
|
// activation in the half-open interval `(after, until]`, in canonical
|
|
// UTC order (ascending). Used by the Autopilot schedule dispatch job to
|
|
// enumerate every plan_time that became due between the last stored
|
|
// occurrence and DB now().
|
|
//
|
|
// The slice is capped at 1024 entries — a safety net against an
|
|
// accidental "every second" cron over a multi-day catch-up window.
|
|
// The scheduler manager additionally caps the returned slice at
|
|
// JobSpec.MaxPlansPerTick.
|
|
func NextOccurrencesUTC(cronExpr, timezone string, after, until time.Time) ([]time.Time, error) {
|
|
sched, loc, err := parseCronSchedule(cronExpr, timezone)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
const hardCap = 1024
|
|
out := make([]time.Time, 0, 8)
|
|
cursor := after.In(loc)
|
|
untilLocal := until.In(loc)
|
|
for len(out) < hardCap {
|
|
next := sched.Next(cursor)
|
|
if next.After(untilLocal) {
|
|
break
|
|
}
|
|
out = append(out, next.UTC())
|
|
cursor = next
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ComputeNextRun evaluates the cron at the app's local now() and backs
|
|
// the display-only autopilot_trigger.next_run_at column for the trigger
|
|
// create/update handlers and the failure monitor. Using the local clock
|
|
// for this display value is deliberate: app/DB clock skew under NTP is far
|
|
// below the column's minute-level granularity, so threading DB time
|
|
// through these UI write paths would buy no user-visible accuracy.
|
|
//
|
|
// The scheduler's post-dispatch advance keeps next_run_at on the same
|
|
// local clock but calls NextOccurrenceAfterUTC anchored at
|
|
// max(now, plan_time) rather than this helper, so a lagging app clock can
|
|
// never re-point the column at the slot that just fired (see
|
|
// scheduler.advancedNextRun / autopilotHandler, MUL-3749).
|
|
//
|
|
// Scheduling decisions are a separate concern and MUST go through
|
|
// NextOccurrencesUTC / NextOccurrenceAfterUTC against DB time instead:
|
|
// dispatch correctness across clock-skewed app instances depends on it.
|
|
func ComputeNextRun(cronExpr, timezone string) (time.Time, error) {
|
|
return NextOccurrenceAfterUTC(cronExpr, timezone, time.Now())
|
|
}
|
|
|
|
// ValidateTimezone returns an error if the timezone string is not recognized.
|
|
func ValidateTimezone(timezone string) error {
|
|
_, err := time.LoadLocation(timezone)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid timezone %q: %w", timezone, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func parseCronSchedule(cronExpr, timezone string) (cron.Schedule, *time.Location, error) {
|
|
sched, err := cronParser.Parse(cronExpr)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("parse cron: %w", err)
|
|
}
|
|
loc, err := time.LoadLocation(timezone)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("invalid timezone %q: %w", timezone, err)
|
|
}
|
|
return sched, loc, nil
|
|
}
|