oficina / calendar Public

Download this file

1	package calendar
2
3	import (
4	"context"
5	"errors"
6	"net/http"
7	"net/http/httptest"
8	"sync"
9	"sync/atomic"
10	"testing"
11	"time"
12
13	"amadan.net/rastrillo/rastrillo/carlos"
14	)
15
16	type scheduleRecorder struct {
17	mu sync.Mutex
18	calls []time.Time
19	fail bool
20	}
21
22	func (r *scheduleRecorder) write(ctx context.Context, at time.Time) error {
23	r.mu.Lock()
24	defer r.mu.Unlock()
25	if r.fail {
26	return context.DeadlineExceeded
27	}
28	r.calls = append(r.calls, at)
29	return nil
30	}
31
32	func (r *scheduleRecorder) snapshot() []time.Time {
33	r.mu.Lock()
34	defer r.mu.Unlock()
35	return append([]time.Time(nil), r.calls...)
36	}
37
38	func testScheduledWork(t *testing.T, query func(context.Context) (time.Time, error)) (*scheduledWork, *scheduleRecorder) {
39	t.Helper()
40	r := &scheduleRecorder{}
41	s := newScheduledWork(query, false, nil)
42	s.retryDelay = time.Millisecond
43	s.capability = func(context.Context) error { return nil }
44	s.armed = func(context.Context) (remoteSchedule, error) {
45	return remoteSchedule{NextAt: s.registeredAt.Unix(), DueAt: s.registeredAt.Unix(), Path: "/jobs/reminders", Kind: "once"}, nil
46	}
47	s.put = func(ctx context.Context, name string, at time.Time, path string) error {
48	if name != scheduledWorkName || path != "/jobs/reminders" {
49	t.Errorf("unexpected registration: %q %q", name, path)
50	}
51	return r.write(ctx, at)
52	}
53	s.remove = func(ctx context.Context, name string) error { return r.write(ctx, time.Time{}) }
54	if err := s.start(); err != nil {
55	t.Fatal(err)
56	}
57	t.Cleanup(s.close)
58	return s, r
59	}
60
61	func TestScheduledWorkKeepsUnchangedRequestsFromResettingRetries(t *testing.T) {
62	due := time.Now().Add(time.Hour).Truncate(time.Second)
63	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
64	for range 3 {
65	if _, err := s.sync(false); err != nil {
66	t.Fatal(err)
67	}
68	}
69	if n := len(r.snapshot()); n != 1 {
70	t.Fatalf("ordinary requests wrote %d registrations, want 1", n)
71	}
72	if _, err := s.sync(true); err != nil {
73	t.Fatal(err)
74	}
75	if n := len(r.snapshot()); n != 2 {
76	t.Fatalf("completed delivery did not rearm the same due time: %d", n)
77	}
78	}
79
80	func TestScheduledWorkRetriesAForcedRearmWithoutAVisitor(t *testing.T) {
81	due := time.Now().Add(time.Hour)
82	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
83	r.mu.Lock()
84	r.fail = true
85	r.mu.Unlock()
86	if _, err := s.sync(true); err == nil {
87	t.Fatal("registration failure was swallowed")
88	}
89	r.mu.Lock()
90	r.fail = false
91	r.mu.Unlock()
92	deadline := time.After(time.Second)
93	for len(r.snapshot()) != 2 {
94	select {
95	case <-deadline:
96	t.Fatal("failed forced registration was never retried")
97	case <-time.After(time.Millisecond):
98	}
99	}
100	if _, err := s.sync(false); err != nil {
101	t.Fatal(err)
102	}
103	if n := len(r.snapshot()); n != 2 {
104	t.Fatalf("successful retry was repeated by ordinary request: %d", n)
105	}
106	}
107
108	func TestScheduledWorkReadsAfterOutOfOrderRequestCompletion(t *testing.T) {
109	var due atomic.Int64
110	due.Store(time.Now().Add(2 * time.Hour).Unix())
111	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
112	entered, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
113	first := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
114	due.Store(time.Now().Add(time.Hour).Unix())
115	close(entered)
116	<-release
117	}))
118	go func() {
119	defer close(finished)
120	first.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/first", nil))
121	}()
122	<-entered
123	earlier := time.Now().Add(time.Minute).Unix()
124	second := s.middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { due.Store(earlier) }))
125	second.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("POST", "/second", nil))
126	close(release)
127	<-finished
128	if _, err := s.sync(false); err != nil {
129	t.Fatal(err)
130	}
131	calls := r.snapshot()
132	if len(calls) != 2 || calls[1].Unix() != earlier {
133	t.Fatalf("stale request replaced the earlier registration: %v", calls)
134	}
135	}
136
137	func TestScheduledWorkSerializesQueryWithRegistration(t *testing.T) {
138	var due atomic.Int64
139	due.Store(time.Now().Add(time.Hour).Unix())
140	entered, release := make(chan struct{}), make(chan struct{})
141	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
142	oldPut := s.put
143	var blocked atomic.Bool
144	s.put = func(ctx context.Context, name string, at time.Time, path string) error {
145	if blocked.CompareAndSwap(false, true) {
146	close(entered)
147	<-release
148	}
149	return oldPut(ctx, name, at, path)
150	}
151	firstDone := make(chan struct{})
152	go func() { defer close(firstDone); _, _ = s.sync(true) }()
153	<-entered
154	earlier := time.Now().Add(time.Minute).Unix()
155	due.Store(earlier)
156	secondDone := make(chan struct{})
157	go func() { defer close(secondDone); _, _ = s.sync(false) }()
158	// The second write must not overtake the first while its control call is held.
159	select {
160	case <-secondDone:
161	t.Fatal("query and registration were not serialized")
162	case <-time.After(10 * time.Millisecond):
163	}
164	close(release)
165	<-firstDone
166	<-secondDone
167	calls := r.snapshot()
168	if calls[len(calls)-1].Unix() != earlier {
169	t.Fatalf("last registration = %v", calls)
170	}
171	}
172
173	func TestScheduledWorkQueryFailureCannotCancelOrForgetWork(t *testing.T) {
174	var broken atomic.Bool
175	due := time.Now().Add(-time.Minute)
176	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) {
177	if broken.Load() {
178	return time.Time{}, context.DeadlineExceeded
179	}
180	return due, nil
181	})
182	broken.Store(true)
183	got, err := s.sync(false)
184	if err == nil || !got.Equal(due) {
185	t.Fatalf("failed query = %v, %v", got, err)
186	}
187	if len(r.snapshot()) != 1 {
188	t.Fatal("failed query cancelled the timer")
189	}
190	}
191
192	func TestScheduledWorkFarFutureAndCancellation(t *testing.T) {
193	var due atomic.Int64
194	due.Store(time.Now().Add(2 * carlos.MaxAhead).Unix())
195	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) {
196	if n := due.Load(); n != 0 {
197	return time.Unix(n, 0), nil
198	}
199	return time.Time{}, nil
200	})
201	got := r.snapshot()[0]
202	if got.After(time.Now().Add(carlos.MaxAhead)) || got.Before(time.Now().Add(carlos.MaxAhead-2*time.Minute)) {
203	t.Fatalf("far future recheck = %v", got)
204	}
205	if _, err := s.sync(false); err != nil {
206	t.Fatal(err)
207	}
208	if len(r.snapshot()) != 1 {
209	t.Fatal("far future registration drifted on an ordinary request")
210	}
211	due.Store(0)
212	if _, err := s.sync(false); err != nil {
213	t.Fatal(err)
214	}
215	if calls := r.snapshot(); len(calls) != 2 || !calls[1].IsZero() {
216	t.Fatalf("empty queue was not cancelled: %v", calls)
217	}
218	}
219
220	func TestScheduledWorkOffPlatformIsDevelopmentOnly(t *testing.T) {
221	t.Setenv("CARLOS_CONTROL_SOCKET", "")
222	for _, development := range []bool{false, true} {
223	s := newScheduledWork(func(context.Context) (time.Time, error) { return time.Time{}, nil }, development, nil)
224	err := s.start()
225	if development {
226	if err != nil {
227	t.Fatal(err)
228	}
229	s.close()
230	} else if !errors.Is(err, carlos.ErrNotOnCarlos) {
231	t.Fatalf("production startup = %v", err)
232	}
233	}
234	}
235
236	func TestScheduledWorkCloseCancelsAndJoinsRegistration(t *testing.T) {
237	s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
238	entered := make(chan struct{})
239	s.mu.Lock()
240	s.put = func(ctx context.Context, name string, at time.Time, path string) error {
241	close(entered)
242	<-ctx.Done()
243	return ctx.Err()
244	}
245	s.mu.Unlock()
246	registered := make(chan struct{})
247	go func() { defer close(registered); _, _ = s.sync(true) }()
248	<-entered
249	closed := make(chan struct{})
250	go func() { s.close(); close(closed) }()
251	select {
252	case <-closed:
253	case <-time.After(time.Second):
254	t.Fatal("Close did not cancel registration")
255	}
256	<-registered
257	select {
258	case <-s.done:
259	default:
260	t.Fatal("retry worker outlived Close")
261	}
262	if _, err := s.sync(true); !errors.Is(err, context.Canceled) {
263	t.Fatalf("registration after Close = %v", err)
264	}
265	}
266
267	func TestScheduledWorkRepairsAnAmbiguousControlWrite(t *testing.T) {
268	var due atomic.Int64
269	original := time.Now().Add(time.Hour).Unix()
270	due.Store(original)
271	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Unix(due.Load(), 0), nil })
272	oldPut := s.put
273	var fail atomic.Bool
274	fail.Store(true)
275	s.put = func(ctx context.Context, name string, at time.Time, path string) error {
276	if err := oldPut(ctx, name, at, path); err != nil {
277	return err
278	}
279	if fail.Load() {
280	return context.DeadlineExceeded
281	}
282	return nil
283	}
284	due.Store(original + 60)
285	if _, err := s.sync(false); err == nil {
286	t.Fatal("lost control response was ignored")
287	}
288	due.Store(original)
289	fail.Store(false)
290	if _, err := s.sync(false); err != nil {
291	t.Fatal(err)
292	}
293	calls := r.snapshot()
294	if len(calls) < 3 || calls[len(calls)-1].Unix() != original {
295	t.Fatalf("cached original hid ambiguous replacement: %v", calls)
296	}
297	}
298