v0.168.0: customer-configurable backup window (Mentési időablak)
ONE setting (window start W) drives every nightly leg at fixed, never-stored offsets: DB dump at W, tier-2 at W+60m, off-box at W+105m (wrap-safe). Precedence settings > controller.yaml db_dump_schedule > 02:30. - scheduler.UpdateDaily: retime a daily job at runtime (no restart) via a per-job buffered resched chan + a select case in runDailyJob. - new pure package internal/backupwindow (LegTimes/GateWindow/EffectiveWindow). - quiesce disk-tier window gate: scheduled cycles run only inside [W+2h,W+6h) with a safety valve (age>cadence+24h runs regardless); manual TriggerNow never gated. Backend.Due now also returns the backup age (from the agent's own /backup/due). - backup page: Mentési időablak card (time input + derived leg/gate rows); POST /backups/window validates -> saves -> UpdateDaily x3 -> flash. Tests: 5 groups, all red-proofed. Agent/cadence//backup/due untouched.
This commit is contained in:
@@ -37,6 +37,10 @@ type Job struct {
|
||||
LastRun time.Time
|
||||
LastErr error
|
||||
Running bool
|
||||
// resched (daily jobs only, buffered cap 1) is the runtime-reschedule signal: UpdateDaily mutates
|
||||
// Schedule under the mutex and non-blocking-sends here so runDailyJob wakes and recomputes its next
|
||||
// run immediately, instead of waiting out the old timer. Immutable after job creation.
|
||||
resched chan struct{}
|
||||
}
|
||||
|
||||
// Scheduler manages periodic and daily jobs.
|
||||
@@ -120,6 +124,7 @@ func (s *Scheduler) Daily(name string, timeStr string, fn JobFunc) {
|
||||
Name: name,
|
||||
Fn: fn,
|
||||
Schedule: timeStr,
|
||||
resched: make(chan struct{}, 1),
|
||||
}
|
||||
s.jobs = append(s.jobs, job)
|
||||
|
||||
@@ -133,6 +138,50 @@ func (s *Scheduler) Daily(name string, timeStr string, fn JobFunc) {
|
||||
}
|
||||
}
|
||||
|
||||
// UpdateDaily changes the schedule of an already-registered daily job at runtime (no restart) and
|
||||
// wakes its goroutine so the new time takes effect at the next scheduling pass. Returns false (and
|
||||
// logs WARN) on an invalid time or an unknown/non-daily job name — the job is left untouched.
|
||||
func (s *Scheduler) UpdateDaily(name, timeStr string) bool {
|
||||
if _, _, err := parseDailyTime(timeStr); err != nil {
|
||||
s.logger.Printf("[WARN] [scheduler] UpdateDaily %s: invalid schedule %q: %v — ignored", name, timeStr, err)
|
||||
return false
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
var job *Job
|
||||
for _, j := range s.jobs {
|
||||
if j.Name == name && j.Schedule != "" {
|
||||
job = j
|
||||
break
|
||||
}
|
||||
}
|
||||
if job == nil {
|
||||
s.mu.Unlock()
|
||||
s.logger.Printf("[WARN] [scheduler] UpdateDaily: no daily job named %q — ignored", name)
|
||||
return false
|
||||
}
|
||||
old := job.Schedule
|
||||
if old == timeStr {
|
||||
s.mu.Unlock()
|
||||
return true // no-op: already at this time
|
||||
}
|
||||
job.Schedule = timeStr
|
||||
ch := job.resched
|
||||
s.mu.Unlock()
|
||||
|
||||
s.logger.Printf("[INFO] [scheduler] Daily job %s rescheduled %s → %s (next run %s)",
|
||||
name, old, timeStr, nextDailyRun(timeStr).Format("2006-01-02 15:04 MST"))
|
||||
// Non-blocking wake: the buffered slot coalesces bursts (three legs updated back-to-back each
|
||||
// signal their own job); if the goroutine hasn't drained yet the recompute already covers this.
|
||||
if ch != nil {
|
||||
select {
|
||||
case ch <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Start begins running all registered jobs. Safe to call only once.
|
||||
func (s *Scheduler) Start(ctx context.Context) {
|
||||
s.mu.Lock()
|
||||
@@ -234,7 +283,13 @@ func (s *Scheduler) runDailyJob(job *Job) {
|
||||
defer s.wg.Done()
|
||||
|
||||
for {
|
||||
nextRun := nextDailyRun(job.Schedule)
|
||||
// Read Schedule under the mutex — UpdateDaily mutates it concurrently. resched is immutable
|
||||
// after creation, so it is safe to read once here.
|
||||
s.mu.Lock()
|
||||
schedule := job.Schedule
|
||||
s.mu.Unlock()
|
||||
|
||||
nextRun := nextDailyRun(schedule)
|
||||
waitDuration := time.Until(nextRun)
|
||||
|
||||
if waitDuration < 0 {
|
||||
@@ -249,6 +304,11 @@ func (s *Scheduler) runDailyJob(job *Job) {
|
||||
timer.Stop()
|
||||
s.dbg("daily job %s: context cancelled, stopping", job.Name)
|
||||
return
|
||||
case <-job.resched:
|
||||
// Runtime reschedule: abandon the current timer and recompute against the new Schedule.
|
||||
timer.Stop()
|
||||
s.dbg("daily job %s: rescheduled — recomputing next run", job.Name)
|
||||
continue
|
||||
case <-timer.C:
|
||||
s.executeJob(job, false)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func discardScheduler() *Scheduler { return New(log.New(io.Discard, "", 0)) }
|
||||
|
||||
func noopJob(context.Context) error { return nil }
|
||||
|
||||
// farFuture returns an HH:MM roughly n hours ahead in Budapest, so a registered daily job parks on a
|
||||
// long timer (never fires during a sub-second test) yet is a valid, distinct schedule to switch to.
|
||||
func farFuture(hoursAhead int) string {
|
||||
return time.Now().In(getBudapestLocation()).Add(time.Duration(hoursAhead) * time.Hour).Format("15:04")
|
||||
}
|
||||
|
||||
// Group C — UpdateDaily mutates the schedule and delivers the reschedule signal. Asserted on a
|
||||
// NOT-started scheduler so no goroutine competes for the buffered signal (race-free).
|
||||
func TestUpdateDaily_MutatesAndSignals(t *testing.T) {
|
||||
s := discardScheduler()
|
||||
s.Daily("db-dump", "02:30", noopJob)
|
||||
|
||||
if !s.UpdateDaily("db-dump", "22:00") {
|
||||
t.Fatal("UpdateDaily returned false for a valid change")
|
||||
}
|
||||
if got := s.jobs[0].Schedule; got != "22:00" {
|
||||
t.Errorf("Schedule = %q, want 22:00", got)
|
||||
}
|
||||
// The buffered reschedule signal must be present (no goroutine drained it).
|
||||
select {
|
||||
case <-s.jobs[0].resched:
|
||||
default:
|
||||
t.Error("no reschedule signal was sent by UpdateDaily")
|
||||
}
|
||||
}
|
||||
|
||||
// Group C — unknown or non-daily job name, and an invalid time, both return false without panicking
|
||||
// and leave everything untouched.
|
||||
func TestUpdateDaily_UnknownAndInvalid(t *testing.T) {
|
||||
s := discardScheduler()
|
||||
s.Daily("db-dump", "02:30", noopJob)
|
||||
|
||||
if s.UpdateDaily("does-not-exist", "03:00") {
|
||||
t.Error("UpdateDaily on an unknown job returned true")
|
||||
}
|
||||
if s.UpdateDaily("db-dump", "25:99") {
|
||||
t.Error("UpdateDaily with an invalid time returned true")
|
||||
}
|
||||
if got := s.jobs[0].Schedule; got != "02:30" {
|
||||
t.Errorf("Schedule was mutated by a rejected update: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Group C — a running daily goroutine picks up the new time on the next pass: it CONSUMES the
|
||||
// reschedule signal from the select and recomputes. Red-proof: remove the `case <-job.resched`
|
||||
// from runDailyJob's select and this fails (the signal stays buffered — never consumed).
|
||||
func TestUpdateDaily_GoroutineConsumesReschedule(t *testing.T) {
|
||||
s := discardScheduler()
|
||||
s.Daily("db-dump", farFuture(6), noopJob) // parked on a ~6h timer
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
s.Start(ctx)
|
||||
|
||||
if !s.UpdateDaily("db-dump", farFuture(8)) {
|
||||
t.Fatal("UpdateDaily returned false")
|
||||
}
|
||||
time.Sleep(150 * time.Millisecond) // let the goroutine wake on resched and recompute
|
||||
cancel()
|
||||
s.Stop() // waits for the goroutine to exit — after this, reading resched is race-free
|
||||
|
||||
select {
|
||||
case <-s.jobs[0].resched:
|
||||
t.Fatal("reschedule signal still buffered — runDailyJob never consumed it (no resched case)")
|
||||
default:
|
||||
// drained by the goroutine → immediacy works
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user