1package scheduler
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9)
10
11func TestEngineDispatchDueReschedulesTask(t *testing.T) {
12 loc := time.UTC
13 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)
14 engine := NewEngine(func() time.Time { return now })
15 created, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "check status", Location: loc})
16 if err != nil {
17 t.Fatalf("Add() error = %v", err)
18 }
19
20 done := make(chan Task, 1)
21 now = now.Add(time.Minute)
22 engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
23 done <- task
24 return nil
25 })
26 select {
27 case task := <-done:
28 if task.ID != created.ID {
29 t.Fatalf("executed task ID = %q, want %q", task.ID, created.ID)
30 }
31 case <-time.After(time.Second):
32 t.Fatal("due task was not executed")
33 }
34
35 eventually(t, func() bool {
36 tasks := engine.List()
37 return len(tasks) == 1 && tasks[0].RunCount == 1
38 })
39 tasks := engine.List()
40 if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {
41 t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)
42 }
43 if !tasks[0].LastRun.Equal(now) || tasks[0].RunCount != 1 {
44 t.Fatalf("task after execution = %#v", tasks[0])
45 }
46}
47
48func TestEngineRecordsExecutionFailure(t *testing.T) {
49 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
50 engine := NewEngine(func() time.Time { return now })
51 task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "run", Location: time.UTC})
52 if err != nil {
53 t.Fatalf("Add() error = %v", err)
54 }
55
56 engine.dispatchDue(context.Background(), now.Add(time.Minute), func(context.Context, Task) error { return errors.New("provider unavailable") })
57 eventually(t, func() bool {
58 got, ok := engine.Get(task.ID)
59 return ok && got.LastError == "provider unavailable"
60 })
61}
62
63func TestEngineCancel(t *testing.T) {
64 engine := NewEngine(nil)
65 task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "@hourly", Message: "run", Location: time.UTC})
66 if err != nil {
67 t.Fatalf("Add() error = %v", err)
68 }
69 if !engine.Cancel(task.ID) {
70 t.Fatal("Cancel() = false, want true")
71 }
72 if _, ok := engine.Get(task.ID); ok {
73 t.Fatal("Get() found cancelled task")
74 }
75}
76
77func TestEngineDispatchesLaterTaskWhileEarlierTaskBlocks(t *testing.T) {
78 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
79 engine := NewEngine(func() time.Time { return now })
80 first := testTask(t, "a", now)
81 engine.tasks[first.ID] = first
82
83 started := make(chan struct{})
84 release := make(chan struct{})
85 secondDone := make(chan struct{})
86 engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
87 switch task.ID {
88 case "a":
89 close(started)
90 <-release
91 case "b":
92 close(secondDone)
93 }
94 return nil
95 })
96 <-started
97 if _, ok := engine.nextRun(); ok {
98 t.Fatal("nextRun() considered an in-flight task due")
99 }
100 second := testTask(t, "b", now)
101 engine.mu.Lock()
102 engine.tasks[second.ID] = second
103 engine.mu.Unlock()
104 engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
105 if task.ID == "b" {
106 close(secondDone)
107 }
108 return nil
109 })
110 select {
111 case <-secondDone:
112 case <-time.After(time.Second):
113 t.Fatal("later due task waited for earlier blocked task")
114 }
115 close(release)
116}
117
118func TestEngineClaimsTaskOnceAcrossConcurrentDispatches(t *testing.T) {
119 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
120 engine := NewEngine(func() time.Time { return now })
121 engine.tasks["due"] = testTask(t, "due", now)
122
123 started := make(chan struct{})
124 release := make(chan struct{})
125 var callsMu sync.Mutex
126 calls := 0
127 executor := func(context.Context, Task) error {
128 callsMu.Lock()
129 calls++
130 callsMu.Unlock()
131 close(started)
132 <-release
133 return nil
134 }
135 var dispatched sync.WaitGroup
136 for range 2 {
137 dispatched.Add(1)
138 go func() {
139 defer dispatched.Done()
140 engine.dispatchDue(context.Background(), now, executor)
141 }()
142 }
143 dispatched.Wait()
144 <-started
145 callsMu.Lock()
146 gotCalls := calls
147 callsMu.Unlock()
148 if gotCalls != 1 {
149 t.Fatalf("executor calls = %d, want 1", gotCalls)
150 }
151 close(release)
152}
153
154func TestEngineCancelQueuedTaskBeforeCallback(t *testing.T) {
155 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
156 engine := NewEngine(func() time.Time { return now })
157 first := testTask(t, "a", now)
158 second := testTask(t, "b", now)
159 engine.tasks[first.ID] = first
160 engine.tasks[second.ID] = second
161 due := engine.takeDue(now)
162 if len(due) != 2 {
163 t.Fatalf("takeDue() returned %d tasks, want 2", len(due))
164 }
165
166 started := make(chan struct{})
167 release := make(chan struct{})
168 var wg sync.WaitGroup
169 wg.Add(1)
170 go func() {
171 defer wg.Done()
172 engine.dispatchTask(context.Background(), due[0], func(context.Context, Task) error {
173 close(started)
174 <-release
175 return nil
176 })
177 }()
178 <-started
179 if !engine.Cancel(second.ID) {
180 t.Fatal("Cancel(second) = false, want true")
181 }
182 calledSecond := false
183 engine.dispatchTask(context.Background(), due[1], func(context.Context, Task) error {
184 calledSecond = true
185 return nil
186 })
187 close(release)
188 wg.Wait()
189 if calledSecond {
190 t.Fatal("cancelled queued task invoked its executor")
191 }
192}
193
194func TestEngineRunSkipsDueTasksWhenContextCancelled(t *testing.T) {
195 now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
196 engine := NewEngine(func() time.Time { return now })
197 engine.tasks["due"] = testTask(t, "due", now)
198 ctx, cancel := context.WithCancel(context.Background())
199 cancel()
200 called := false
201 if err := engine.Run(ctx, func(context.Context, Task) error {
202 called = true
203 return nil
204 }); err != nil {
205 t.Fatalf("Run() error = %v", err)
206 }
207 if called {
208 t.Fatal("Run() invoked executor after cancellation")
209 }
210}
211
212func testTask(t *testing.T, id string, now time.Time) Task {
213 t.Helper()
214 schedule, err := ParseCron("* * * * *", time.UTC)
215 if err != nil {
216 t.Fatal(err)
217 }
218 return Task{ID: id, SessionID: id, Cron: "* * * * *", Message: id, NextRun: now, schedule: schedule}
219}
220
221func eventually(t *testing.T, condition func() bool) {
222 t.Helper()
223 deadline := time.Now().Add(time.Second)
224 for time.Now().Before(deadline) {
225 if condition() {
226 return
227 }
228 time.Sleep(time.Millisecond)
229 }
230 t.Fatal("condition did not become true before deadline")
231}