oficina / calendar Public

Download this file

1	package calendar
2
3	import (
4	"context"
5	"sync"
6	"sync/atomic"
7	"testing"
8	"time"
9	)
10
11	type remoteTimerFixture struct {
12	mu sync.Mutex
13	timer remoteSchedule
14	writes int
15	}
16
17	func restartRegistrar(t *testing.T, due time.Time, clock *atomic.Int64, remote *remoteTimerFixture) *scheduledWork {
18	t.Helper()
19	s := newScheduledWork(func(context.Context) (time.Time, error) { return due, nil }, false, nil)
20	s.now = func() time.Time { return time.Unix(clock.Load(), 0) }
21	s.capability = func(context.Context) error { return nil }
22	s.armed = func(context.Context) (remoteSchedule, error) {
23	remote.mu.Lock()
24	defer remote.mu.Unlock()
25	return remote.timer, nil
26	}
27	s.put = func(_ context.Context, _ string, at time.Time, _ string) error {
28	remote.mu.Lock()
29	defer remote.mu.Unlock()
30	remote.timer = remoteSchedule{NextAt: at.Unix(), DueAt: at.Unix(), Kind: "once", Path: "/jobs/reminders"}
31	remote.writes++
32	return nil
33	}
34	if err := s.start(); err != nil {
35	t.Fatal(err)
36	}
37	t.Cleanup(s.close)
38	return s
39	}
40
41	func TestDrainBackoffSurvivesRestartThroughPersistedLateness(t *testing.T) {
42	var clock atomic.Int64
43	clock.Store(time.Now().Unix())
44	due := time.Unix(clock.Load(), 0).Add(-4 * time.Minute)
45	remote := &remoteTimerFixture{}
46	first := restartRegistrar(t, due, &clock, remote)
47	if err := first.finishDrain(first.now(), 0, context.DeadlineExceeded); err != nil {
48	t.Fatal(err)
49	}
50	first.close()
51	original := remote.timer.NextAt
52	writes := remote.writes
53	clock.Add(30)
54	second := restartRegistrar(t, due, &clock, remote)
55	if remote.writes != writes || remote.timer.NextAt != original {
56	t.Fatal("boot replaced existing future backoff")
57	}
58	if got := second.probe(); got.Unix() != original {
59	t.Fatalf("boot probe bypassed backoff: %v", got)
60	}
61	second.close()
62	clock.Add(12 * 60)
63	third := restartRegistrar(t, due, &clock, remote)
64	if remote.writes != writes {
65	t.Fatal("boot replaced the delivery currently waking the process")
66	}
67	if err := third.finishDrain(third.now(), 0, context.DeadlineExceeded); err != nil {
68	t.Fatal(err)
69	}
70	if remote.timer.NextAt < clock.Load()+4*60 {
71	t.Fatalf("failure restarted at one minute: %d", remote.timer.NextAt-clock.Load())
72	}
73	}
74
75	func TestSuccessfulDrainWithoutProgressBacksOffButSweepProgressDoesNot(t *testing.T) {
76	due := time.Now().Add(-time.Minute)
77	s, r := testScheduledWork(t, func(context.Context) (time.Time, error) { return due, nil })
78	now := s.now()
79	if err := s.finishDrain(now, 0, nil); err != nil {
80	t.Fatal(err)
81	}
82	calls := r.snapshot()
83	if calls[len(calls)-1].Before(now.Add(time.Minute)) {
84	t.Fatalf("no-progress success immediately rearmed: %v", calls)
85	}
86	if err := s.finishDrain(s.now(), 500, nil); err != nil {
87	t.Fatal(err)
88	}
89	calls = r.snapshot()
90	if calls[len(calls)-1].After(time.Now()) {
91	t.Fatal("real sweep progress delayed remaining backlog")
92	}
93	}
94
95	func TestLateCommittedReplacementIsRepairedWithoutResettingPlatformRetries(t *testing.T) {
96	var clock atomic.Int64
97	clock.Store(time.Now().Unix())
98	due := time.Unix(clock.Load(), 0).Add(time.Hour)
99	remote := &remoteTimerFixture{}
100	s := restartRegistrar(t, due, &clock, remote)
101	remote.mu.Lock()
102	remote.timer.NextAt = due.Add(time.Hour).Unix()
103	remote.timer.DueAt = remote.timer.NextAt
104	remote.mu.Unlock()
105	if _, err := s.synchronize(false, true); err != nil {
106	t.Fatal(err)
107	}
108	if remote.timer.NextAt != due.Unix() || remote.writes != 2 {
109	t.Fatalf("late stale write survived inspection: %+v", remote)
110	}
111	clock.Store(due.Unix() + 60)
112	remote.mu.Lock()
113	remote.timer.NextAt = clock.Load() + 8*60
114	remote.timer.Attempts = 4
115	remote.mu.Unlock()
116	if _, err := s.synchronize(false, true); err != nil {
117	t.Fatal(err)
118	}
119	if remote.writes != 2 {
120	t.Fatal("legitimate platform retry was reset")
121	}
122	}
123
124	func TestNextDueBusyRegistrationReturnsImmediately(t *testing.T) {
125	s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
126	s.mu.Lock()
127	started := time.Now()
128	done := make(chan time.Time, 1)
129	go func() { done <- s.probe() }()
130	select {
131	case due := <-done:
132	if due.Before(started) || due.After(time.Now()) {
133	t.Fatal(due)
134	}
135	case <-time.After(100 * time.Millisecond):
136	s.mu.Unlock()
137	t.Fatal("probe waited for registration mutex")
138	}
139	s.mu.Unlock()
140	}
141
142	func TestNextDueHasOneOverallBudgetForSequentialControlCalls(t *testing.T) {
143	s, _ := testScheduledWork(t, func(context.Context) (time.Time, error) { return time.Now().Add(time.Hour), nil })
144	s.mu.Lock()
145	s.force = true
146	s.capability = func(ctx context.Context) error {
147	select {
148	case <-ctx.Done():
149	return ctx.Err()
150	case <-time.After(1500 * time.Millisecond):
151	return nil
152	}
153	}
154	s.put = func(ctx context.Context, _ string, _ time.Time, _ string) error { <-ctx.Done(); return ctx.Err() }
155	s.query = func(ctx context.Context) (time.Time, error) {
156	select {
157	case <-ctx.Done():
158	return time.Time{}, ctx.Err()
159	case <-time.After(1500 * time.Millisecond):
160	return time.Now().Add(time.Hour), nil
161	}
162	}
163	s.mu.Unlock()
164	started := time.Now()
165	due := s.probe()
166	if elapsed := time.Since(started); elapsed >= 4500*time.Millisecond {
167	t.Fatalf("probe exceeded pre-stop budget: %v", elapsed)
168	}
169	if due.IsZero() || due.After(time.Now()) {
170	t.Fatalf("failed probe did not preserve wake: %v", due)
171	}
172	}
173