From 093b2bd3676d6bb061afc798e500bc1380b5ddf9 Mon Sep 17 00:00:00 2001 From: ji ruo <79234113+RuoJi6@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:09:17 +0800 Subject: [PATCH] fix: preserve manual task starts outside schedule windows --- db/schema.sql | 1 + server/calendar_schedule.go | 59 +++++- server/calendar_schedule_manual_test.go | 250 ++++++++++++++++++++++++ server/goals.go | 17 +- server/manager.go | 11 +- 5 files changed, 334 insertions(+), 4 deletions(-) create mode 100644 server/calendar_schedule_manual_test.go diff --git a/db/schema.sql b/db/schema.sql index c8b68f6..cab9a5c 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -1071,6 +1071,7 @@ CREATE TABLE IF NOT EXISTS task_schedule_tasks ( PRIMARY KEY (schedule_id, task_id) ); ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_paused BOOLEAN NOT NULL DEFAULT false; +ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_manual_until TIMESTAMPTZ; CREATE INDEX IF NOT EXISTS idx_task_schedule_tasks_task ON task_schedule_tasks(task_id); ALTER TABLE task_schedule_tasks ADD COLUMN IF NOT EXISTS paused_by_schedule BOOLEAN NOT NULL DEFAULT false; diff --git a/server/calendar_schedule.go b/server/calendar_schedule.go index 5992f6e..02a85a3 100644 --- a/server/calendar_schedule.go +++ b/server/calendar_schedule.go @@ -154,14 +154,51 @@ func evaluateRuntimeScheduleWindow(item *db.TaskSchedule, now, startedAt time.Ti return w } +// taskManualScheduleEnd allows only the selected task to run until the next +// enabled calendar's end. The normal union of windows still governs overlapping +// plans after that boundary. Open-ended or disabled plans impose no deadline. +func (s *Server) taskManualScheduleEnd(taskID string, now time.Time) (*time.Time, error) { + id, err := strconv.ParseInt(taskID, 10, 64) + if err != nil { + return nil, err + } + items, err := s.m.pg.ListTaskSchedules() + if err != nil { + return nil, err + } + var until *time.Time + for _, item := range items { + for _, boundID := range item.TaskIDs { + if boundID != id { + continue + } + if until == nil { + unbounded := time.Date(9999, 12, 31, 0, 0, 0, 0, time.UTC) + until = &unbounded + } + if item.Enabled && item.Status != "completed" { + w := evaluateScheduleWindow(item, now) + if w.nextEnd != nil && w.nextEnd.After(now) && w.nextEnd.Before(*until) { + until = w.nextEnd + } + } + break + } + } + return until, nil +} + func (sc *Scheduler) fireTaskSchedules() { + sc.fireTaskSchedulesAt(time.Now()) +} + +func (sc *Scheduler) fireTaskSchedulesAt(now time.Time) { sc.s.scheduleMu.Lock() defer sc.s.scheduleMu.Unlock() items, err := sc.pg.ListTaskSchedules() if err != nil { return } - now := time.Now() open := map[int64]bool{} windows := map[int64]scheduleWindow{} for _, item := range items { @@ -173,6 +210,26 @@ func (sc *Scheduler) fireTaskSchedules() { } } } + // Persisted task-level manual starts survive ticks, FIFO promotion and + // restarts. Their deadline is the next calendar end, even if no scheduler + // tick ran inside that window. + rows, err := sc.pg.Query(`SELECT id FROM tasks WHERE schedule_manual_until > $1 AND deleted_at IS NULL`, now) + if err != nil { + return + } + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + rows.Close() + return + } + open[id] = true + } + err = rows.Err() + rows.Close() + if err != nil { + return + } // Evaluate the union first, then control each task at most once per tick. seen := map[int64]bool{} for _, item := range items { diff --git a/server/calendar_schedule_manual_test.go b/server/calendar_schedule_manual_test.go new file mode 100644 index 0000000..c8fc452 --- /dev/null +++ b/server/calendar_schedule_manual_test.go @@ -0,0 +1,250 @@ +package server + +import ( + "bytes" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strconv" + "testing" + "time" + + "github.com/Autumn-27/artex/db" +) + +func manualScheduleTestServer(t *testing.T) (*Server, *Scheduler) { + t.Helper() + if os.Getenv("ARTEX_PG_DSN") == "" { + t.Skip("requires isolated ARTEX_PG_DSN") + } + m, err := NewManager(t.TempDir(), "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = m.Close() }) + t.Cleanup(restoreConcurrencySetting(t, m)) + if err := m.SetConcurrency(false, defaultConcurrencyLimit); err != nil { + t.Fatal(err) + } + s := newAdmissionTestServer(m, nil) + sc := &Scheduler{s: s, pg: m.pg, startedAt: calendarTestTime("2026-10-19T08:00:00+08:00")} + return s, sc +} + +func manualScheduleTask(t *testing.T, s *Server) *Task { + t.Helper() + task, err := s.m.CreateTask("manual schedule test", "test admission", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + // Avoid bootstrapping real agent work; this is an existing paused task. + task.updateLifecycle(func(state *taskLifecycleState) { state.FirstRunAt = 1 }) + t.Cleanup(func() { _, _ = s.m.DeleteTask(task.ID, DeleteTaskOptions{}) }) + return task +} + +func manualSchedulePlan(t *testing.T, s *Server, tasks ...*Task) *db.TaskSchedule { + t.Helper() + ids := []int64{} + for _, task := range tasks { + id, _ := strconv.ParseInt(task.ID, 10, 64) + ids = append(ids, id) + } + plan, err := s.m.pg.CreateTaskSchedule(db.TaskScheduleInput{Name: "18:00 calendar", Enabled: true, ScheduleType: "weekly", TimezoneMode: "beijing", Weekdays: []int{1, 2, 3, 4, 5}, StartTime: "18:00", EndTime: "23:00", TaskIDs: ids}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.m.pg.DeleteTaskSchedule(plan.ID) }) + return plan +} + +// Admission uses the real clock. Verify it wrote a live override, then freeze +// just its deadline so calendar ticks can exercise exact boundaries on any date. +func freezeManualScheduleDeadline(t *testing.T, s *Server, task *Task) { + t.Helper() + var until time.Time + if err := s.m.pg.QueryRow(`SELECT schedule_manual_until FROM tasks WHERE id=$1`, task.ID).Scan(&until); err != nil { + t.Fatal(err) + } + if !until.After(time.Now()) { + t.Fatalf("manual deadline already expired: %v", until) + } + if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_manual_until=$2 WHERE id=$1`, task.ID, calendarTestTime("2026-10-19T23:00:00+08:00")); err != nil { + t.Fatal(err) + } +} + +func TestCalendarManualTaskResumeSurvivesTicks(t *testing.T) { + for _, batch := range []bool{false, true} { + t.Run(fmt.Sprintf("batch=%v", batch), func(t *testing.T) { + s, sc := manualScheduleTestServer(t) + task, other := manualScheduleTask(t, s), manualScheduleTask(t, s) + manualSchedulePlan(t, s, task, other) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:00+08:00")) + if !task.lifecycleSnapshot().Paused || !other.lifecycleSnapshot().Paused { + t.Fatal("outside window must initially pause both tasks") + } + body := `{"action":"resume"}` + if batch { + body = fmt.Sprintf(`{"action":"resume","task_ids":[%q]}`, task.ID) + } + req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body)) + req.SetPathValue("id", task.ID) + rec := httptest.NewRecorder() + if batch { + s.controlTasksBatch(rec, req) + } else { + s.control(rec, req) + } + if rec.Code != http.StatusOK { + t.Fatalf("resume: %d %s", rec.Code, rec.Body.String()) + } + freezeManualScheduleDeadline(t, s, task) + for _, at := range []string{"2026-10-19T17:00:05+08:00", "2026-10-19T17:00:10+08:00"} { + sc.fireTaskSchedulesAt(calendarTestTime(at)) + if task.lifecycleSnapshot().Paused { + t.Fatal("task-page manual resume was paused again by calendar") + } + if !other.lifecycleSnapshot().Paused { + t.Fatal("manual resume started another task in the same plan") + } + } + // A fresh scheduler also reads the committed override. + sc = &Scheduler{s: s, pg: s.m.pg, startedAt: calendarTestTime("2026-10-19T17:01:00+08:00")} + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:01:05+08:00")) + if task.lifecycleSnapshot().Paused { + t.Fatal("scheduler restart lost manual resume") + } + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T18:00:05+08:00")) + if task.lifecycleSnapshot().Paused || other.lifecycleSnapshot().Paused { + t.Fatal("regular window did not open") + } + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T23:00:00+08:00")) + if !task.lifecycleSnapshot().Paused || !other.lifecycleSnapshot().Paused { + t.Fatal("manual override bypassed regular end boundary") + } + }) + } +} + +func TestCalendarManualResumePreservesFIFO(t *testing.T) { + s, sc := manualScheduleTestServer(t) + holder, task := manualScheduleTask(t, s), manualScheduleTask(t, s) + manualSchedulePlan(t, s, task) + if err := s.m.SetConcurrency(true, 1); err != nil { + t.Fatal(err) + } + s.engine.started.Store(holder.ID, true) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:00+08:00")) + result, err := s.applyTaskControl(task, "resume") + if err != nil || !result.Queued { + t.Fatalf("resume should queue: %+v %v", result, err) + } + freezeManualScheduleDeadline(t, s, task) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:05+08:00")) + state := task.lifecycleSnapshot() + if state.Paused || !state.Queued { + t.Fatalf("calendar removed manually resumed task from FIFO: %+v", state) + } + // Promotion uses ordinary admission, which must preserve the override. + if err := s.m.ApplyTaskPause(holder.ID); err != nil { + t.Fatal(err) + } + s.reconcileConcurrency() + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:10+08:00")) + state = task.lifecycleSnapshot() + if state.Paused || state.Queued { + t.Fatalf("promotion lost manual resume: %+v", state) + } + if _, err := s.applyTaskControl(task, "pause"); err != nil { + t.Fatal(err) + } + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T18:00:05+08:00")) + if !task.lifecycleSnapshot().Paused { + t.Fatal("calendar lifted explicit user pause") + } +} + +func TestCalendarManualResumeFailureDoesNotOverride(t *testing.T) { + s, sc := manualScheduleTestServer(t) + task := manualScheduleTask(t, s) + manualSchedulePlan(t, s, task) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:00+08:00")) + if _, err := s.m.pg.Exec(`UPDATE tasks SET status='done' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + if _, err := s.applyTaskControl(task, "resume"); err == nil { + t.Fatal("stale resume must fail") + } + var manual bool + if err := s.m.pg.QueryRow(`SELECT schedule_manual_until IS NOT NULL FROM tasks WHERE id=$1`, task.ID).Scan(&manual); err != nil { + t.Fatal(err) + } + if manual { + t.Fatal("failed resume committed a schedule override") + } +} + +func TestCalendarScheduleRunNowStillWorks(t *testing.T) { + s, sc := manualScheduleTestServer(t) + task := manualScheduleTask(t, s) + plan := manualSchedulePlan(t, s, task) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:00+08:00")) + req := httptest.NewRequest(http.MethodPost, "/", nil) + req.SetPathValue("id", strconv.FormatInt(plan.ID, 10)) + rec := httptest.NewRecorder() + s.runScheduleNow(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("run now: %d %s", rec.Code, rec.Body.String()) + } + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:05+08:00")) + if task.lifecycleSnapshot().Paused { + t.Fatal("schedule-page run now regressed") + } +} + +func TestCalendarManualDeadlineAcrossRestart(t *testing.T) { + s, sc := manualScheduleTestServer(t) + task := manualScheduleTask(t, s) + manualSchedulePlan(t, s, task) + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T17:00:00+08:00")) + if _, err := s.applyTaskControl(task, "resume"); err != nil { + t.Fatal(err) + } + freezeManualScheduleDeadline(t, s, task) + // The process was down across the entire window; no active tick consumed it. + sc = &Scheduler{s: s, pg: s.m.pg, startedAt: calendarTestTime("2026-10-19T23:05:00+08:00")} + sc.fireTaskSchedulesAt(calendarTestTime("2026-10-19T23:05:05+08:00")) + if !task.lifecycleSnapshot().Paused { + t.Fatal("restart across window retained expired manual override") + } +} + +func TestCalendarManualDeadlineUsesNextEnd(t *testing.T) { + s, _ := manualScheduleTestServer(t) + task := manualScheduleTask(t, s) + plan := manualSchedulePlan(t, s, task) + for _, tc := range []struct{ at, want string }{ + {"2026-10-19T17:00:00+08:00", "2026-10-19T23:00:00+08:00"}, + {"2026-10-19T19:00:00+08:00", "2026-10-19T23:00:00+08:00"}, + {"2026-10-19T23:01:00+08:00", "2026-10-20T23:00:00+08:00"}, + } { + until, err := s.taskManualScheduleEnd(task.ID, calendarTestTime(tc.at)) + if err != nil || until == nil || !until.Equal(calendarTestTime(tc.want)) { + t.Fatalf("at %s: deadline=%v err=%v", tc.at, until, err) + } + } + if err := s.m.pg.SetTaskScheduleEnabled(plan.ID, false); err != nil { + t.Fatal(err) + } + until, err := s.taskManualScheduleEnd(task.ID, calendarTestTime("2026-10-19T17:00:00+08:00")) + if err != nil || until == nil || until.Year() != 9999 { + t.Fatalf("disabled calendar deadline=%v err=%v", until, err) + } + plain := manualScheduleTask(t, s) + until, err = s.taskManualScheduleEnd(plain.ID, calendarTestTime("2026-10-19T17:00:00+08:00")) + if err != nil || until != nil { + t.Fatalf("unscheduled task deadline=%v err=%v", until, err) + } +} diff --git a/server/goals.go b/server/goals.go index 513502e..4603149 100644 --- a/server/goals.go +++ b/server/goals.go @@ -7,6 +7,7 @@ import ( "sort" "strconv" "strings" + "time" "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/db" @@ -114,6 +115,13 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch if t == nil { return false, fmt.Errorf("task not found") } + manualSchedule := requirePaused && !scheduled + if manualSchedule { + // Match the calendar tick's lock order: scheduleMu before concMu. The + // override must commit before another tick can decide to pause this task. + s.scheduleMu.Lock() + defer s.scheduleMu.Unlock() + } if mode != "bootstrap" { mode = "resume" } @@ -149,6 +157,13 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch } mode = s.resumeAdmissionMode(t) } + var manualUntil *time.Time + if manualSchedule { + manualUntil, err = s.taskManualScheduleEnd(t.ID, time.Now()) + if err != nil { + return false, err + } + } wasTerminal := isTerminalStatus(lifecycle.Status) wasPaused := lifecycle.Paused || s.engine.IsPaused(t.ID) wasQueued := lifecycle.Queued @@ -197,7 +212,7 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch if wasTerminal { status = "running" } - if err := s.m.ApplyTaskAdmission(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued); err != nil { + if err := s.m.applyTaskAdmission(t.ID, lifecycle.Status, status, shouldQueue, mode, wasQueued, manualUntil); err != nil { if !engineWasPaused && !wasPaused && !wasQueued { s.engine.Resume(t) } diff --git a/server/manager.go b/server/manager.go index 11a4dd1..09fdcd3 100644 --- a/server/manager.go +++ b/server/manager.go @@ -1153,6 +1153,12 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error { // admission keeps its FIFO timestamp; a task that was explicitly paused and is // now re-queued receives a fresh tail position. func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool) error { + return m.applyTaskAdmission(id, expectedStatus, status, queued, mode, preservePosition, nil) +} + +// Manual task control records its schedule override in the admission commit, so +// failed resumes cannot bypass a calendar and FIFO promotion preserves it. +func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued bool, mode string, preservePosition bool, manualUntil *time.Time) error { m.taskStateMu.Lock() defer m.taskStateMu.Unlock() n, err := strconv.ParseInt(id, 10, 64) @@ -1176,6 +1182,7 @@ func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued b ELSE NULL END, paused=false, schedule_paused=false, + schedule_manual_until=COALESCE($7::timestamptz, schedule_manual_until), queued=$3, queued_at=CASE WHEN NOT $3 THEN NULL @@ -1196,7 +1203,7 @@ func (m *Manager) ApplyTaskAdmission(id, expectedStatus, status string, queued b ELSE deadline_at END WHERE id=$1 AND deleted_at IS NULL AND status=$6 - RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at`, n, status, queued, mode, preservePosition, expectedStatus). + RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil). Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt) if errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus) @@ -1249,7 +1256,7 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error { } var mode string err = m.pg.QueryRow(`UPDATE tasks - SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2 + SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout') RETURNING COALESCE(queue_mode,'')`, n, scheduled).Scan(&mode)