amadan

oficina / calendar Public

Clone 
git clone https://amadan.net/oficina/calendar

Plain git — no account needed to clone.

Download

Download this file

1package calendar
2
3import (
4 "context"
5 "errors"
6 "net/http"
7 "net/http/httptest"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12)
13
14func TestScheduledWorkTransientBootFailureServesAndRecovers(t *testing.T) {
15 for _, queryFailure := range []bool{false, true} {
16 t.Run(map[bool]string{false: "control", true: "query"}[queryFailure], func(t *testing.T) {
17 var fail atomic.Bool
18 fail.Store(true)
19 due := time.Now().Add(time.Hour)
20 s := newScheduledWork(func(context.Context) (time.Time, error) {
21 if queryFailure && fail.Load() {
22 return time.Time{}, context.DeadlineExceeded
23 }
24 return due, nil
25 }, false, nil)
26 s.retryDelay = time.Millisecond
27 s.capability = func(context.Context) error { return nil }
28 s.armed = func(context.Context) (remoteSchedule, error) { return remoteSchedule{}, nil }
29 written := make(chan struct{}, 1)
30 s.put = func(context.Context, string, time.Time, string) error {
31 if fail.Load() {
32 return context.DeadlineExceeded
33 }
34 select {
35 case written <- struct{}{}:
36 default:
37 }
38 return nil
39 }
40 if err := s.start(); err != nil {
41 t.Fatal(err)
42 }
43 t.Cleanup(s.close)
44 fail.Store(false)
45 select {
46 case <-written:
47 case <-time.After(time.Second):
48 t.Fatal("transient boot error never recovered")
49 }
50 })
51 }
52}
53
54func TestScheduledWorkCloseBeforeStartAndAfterPermanentFailure(t *testing.T) {
55 for _, start := range []bool{false, true} {
56 s := newScheduledWork(func(context.Context) (time.Time, error) { return time.Now(), nil }, false, nil)
57 s.capability = func(context.Context) error { return unsupportedScheduling{} }
58 if start {
59 if err := s.start(); err == nil {
60 t.Fatal("unsupported platform accepted")
61 }
62 }
63 done := make(chan struct{})
64 go func() { s.close(); close(done) }()
65 select {
66 case <-done:
67 case <-time.After(time.Second):
68 t.Fatal("Close hung")
69 }
70 }
71}
72
73func TestScheduledWorkDrainFailuresBackOffEveryRegistrationPath(t *testing.T) {
74 now := time.Now().Truncate(time.Second)
75 due := now.Add(-time.Minute)
76 s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
77 s.mu.Lock()
78 s.now = func() time.Time { return now }
79 s.mu.Unlock()
80 for _, minutes := range []int{1, 2, 4, 8, 15, 15} {
81 if err := s.finishDrain(s.now(), 0, context.DeadlineExceeded); err != nil {
82 t.Fatal(err)
83 }
84 expected := now.Add(time.Duration(minutes) * time.Minute)
85 for _, inspect := range []bool{false, true} {
86 got, err := s.synchronize(false, inspect)
87 if err != nil || !got.Equal(expected) {
88 t.Fatalf("failure floor bypassed: %v %v, want %v", got, err, expected)
89 }
90 }
91 calls := r.snapshot()
92 if !calls[len(calls)-1].Equal(expected) {
93 t.Fatalf("backoff registration = %v, want %v", calls, expected)
94 }
95 }
96 if err := s.finishDrain(s.now(), 1, nil); err != nil {
97 t.Fatal(err)
98 }
99 calls := r.snapshot()
100 if !calls[len(calls)-1].Equal(now) {
101 t.Fatalf("success did not reset backoff: %v", calls)
102 }
103}
104
105func TestScheduledWorkEpochIsDueWorkNotCancellation(t *testing.T) {
106 s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(0, 0), nil })
107 calls := r.snapshot()
108 if calls[0].Unix() <= 0 || calls[0].After(time.Now()) {
109 t.Fatalf("epoch disarmed the timer: %v", calls)
110 }
111 if _, err := s.sync(false); err != nil {
112 t.Fatal(err)
113 }
114 if len(r.snapshot()) != 1 {
115 t.Fatal("clamping defeated unchanged registration cache")
116 }
117}
118
119func TestScheduledWorkPublicRequestsDoNotWaitForControlSocket(t *testing.T) {
120 var due atomic.Int64
121 due.Store(time.Now().Add(time.Hour).Unix())
122 s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
123 entered := make(chan struct{}, 1)
124 s.mu.Lock()
125 s.put = func(ctx context.Context, _ string, _ time.Time, _ string) error {
126 select {
127 case entered <- struct{}{}:
128 default:
129 }
130 <-ctx.Done()
131 return ctx.Err()
132 }
133 s.mu.Unlock()
134 due.Add(1)
135 s.request()
136 select {
137 case <-entered:
138 case <-time.After(time.Second):
139 t.Fatal("worker did not start control request")
140 }
141 h := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) }))
142 start := time.Now()
143 var wg sync.WaitGroup
144 for range 50 {
145 wg.Add(1)
146 go func() {
147 defer wg.Done()
148 h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/book", nil))
149 }()
150 }
151 done := make(chan struct{})
152 go func() { wg.Wait(); close(done) }()
153 select {
154 case <-done:
155 case <-time.After(500 * time.Millisecond):
156 t.Fatal("public traffic queued behind registration")
157 }
158 if time.Since(start) > 500*time.Millisecond {
159 t.Fatal("public latency followed the control timeout")
160 }
161}
162
163func TestScheduledWorkReconcilesConsumedTimersWithoutResettingArmedRows(t *testing.T) {
164 var armed atomic.Bool
165 armed.Store(true)
166 due := time.Now().Add(-time.Minute)
167 s := newScheduledWork(func(context.Context) (time.Time, error) { return due, nil }, false, nil)
168 s.reconcileEvery = 5 * time.Millisecond
169 s.capability = func(context.Context) error { return nil }
170 s.armed = func(context.Context) (remoteSchedule, error) { return remoteSchedule{}, nil }
171 checks := make(chan struct{}, 20)
172 s.armed = func(context.Context) (remoteSchedule, error) {
173 select {
174 case checks <- struct{}{}:
175 default:
176 }
177 if !armed.Load() {
178 return remoteSchedule{}, nil
179 }
180 return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
181 }
182 var writes atomic.Int64
183 s.put = func(context.Context, string, time.Time, string) error { writes.Add(1); armed.Store(true); return nil }
184 if err := s.start(); err != nil {
185 t.Fatal(err)
186 }
187 t.Cleanup(s.close)
188 for range 3 {
189 select {
190 case <-checks:
191 case <-time.After(time.Second):
192 t.Fatal("no periodic reconciliation")
193 }
194 }
195 if writes.Load() != 1 {
196 t.Fatal("armed or in-flight row was rearmed")
197 }
198 armed.Store(false)
199 deadline := time.After(time.Second)
200 for writes.Load() < 2 {
201 select {
202 case <-deadline:
203 t.Fatal("consumed row remained cached")
204 case <-time.After(time.Millisecond):
205 }
206 }
207 if got, err := s.synchronize(false, true); err != nil || !got.Equal(due) {
208 t.Fatalf("repaired due = %v %v", got, err)
209 }
210 if writes.Load() != 2 {
211 t.Fatal("healthy replacement registration was reset")
212 }
213}
214
215func TestScheduledWorkDetectsRollbackBeforeAnotherWrite(t *testing.T) {
216 var due atomic.Int64
217 due.Store(time.Now().Add(time.Hour).Unix())
218 s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
219 s.mu.Lock()
220 s.capability = func(context.Context) error { return unsupportedScheduling{} }
221 s.mu.Unlock()
222 due.Add(60)
223 _, err := s.sync(false)
224 var unsupported unsupportedScheduling
225 if !errors.As(err, &unsupported) {
226 t.Fatalf("rollback was not distinguished: %v", err)
227 }
228 if len(r.snapshot()) != 1 {
229 t.Fatal("wrote unsafe rearm after platform rollback")
230 }
231}
232
233func TestScheduledWorkInspectionFailureDoesNotResetAnArmedRetry(t *testing.T) {
234 due := time.Now().Add(time.Hour)
235 s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
236 var fail atomic.Bool
237 fail.Store(true)
238 s.mu.Lock()
239 s.armed = func(context.Context) (remoteSchedule, error) {
240 if fail.Load() {
241 return remoteSchedule{}, context.DeadlineExceeded
242 }
243 return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
244 }
245 s.mu.Unlock()
246 if _, err := s.synchronize(false, true); err == nil {
247 t.Fatal("inspection failure swallowed")
248 }
249 fail.Store(false)
250 if _, err := s.synchronize(false, true); err != nil {
251 t.Fatal(err)
252 }
253 if len(r.snapshot()) != 1 {
254 t.Fatal("inspection retry replaced healthy armed row")
255 }
256}
257