1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package client_test
4
5import (
6 "context"
7 "errors"
8 "net/http"
9 "net/http/httptest"
10 "sync"
11 "testing"
12 "time"
13
14 "connectrpc.com/connect"
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17
18 "bnema.dev/ermite-api/gen/ermite"
19 "bnema.dev/ermite/internal/adapters/client"
20)
21
22func pushed(id string) *ermite.WatchResponse {
23 return &ermite.WatchResponse{Event: &ermite.Event{Id: id, Kind: &ermite.Event_Pushed{Pushed: &ermite.Pushed{Repo: "r"}}}}
24}
25
26// sleeps records the waits instead of sleeping.
27type sleeps struct {
28 mu sync.Mutex
29 d []time.Duration
30}
31
32func (s *sleeps) sleep(ctx context.Context, d time.Duration) error {
33 s.mu.Lock()
34 s.d = append(s.d, d)
35 s.mu.Unlock()
36 return ctx.Err()
37}
38
39func TestWatchResumesAfterLastEventAndBacksOff(t *testing.T) {
40 ctx, cancel := context.WithCancel(t.Context())
41 defer cancel()
42 var mu sync.Mutex
43 var requests []*ermite.WatchRequest
44 f := &fake{}
45 f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
46 mu.Lock()
47 requests = append(requests, req.Msg)
48 n := len(requests)
49 mu.Unlock()
50 switch n {
51 case 1: // two events, then the connection ends
52 _ = st.Send(pushed("e1"))
53 _ = st.Send(pushed("e2"))
54 return connect.NewError(connect.CodeUnavailable, errors.New("shutting down"))
55 case 2: // fails at once, twice
56 return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
57 case 3:
58 return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
59 default: // back: one more event, then the user stops
60 _ = st.Send(pushed("e3"))
61 <-ctx.Done()
62 return nil
63 }
64 }
65 c := start(t, f, client.Config{})
66
67 var got []string
68 var notices []string
69 sl := &sleeps{}
70 err := c.Watch(ctx, client.WatchOptions{
71 Topics: []string{"home"},
72 Backoff: client.Backoff{Min: time.Second, Max: 3 * time.Second},
73 Sleep: sl.sleep,
74 OnNotice: func(m string) { notices = append(notices, m) },
75 }, func(ev *ermite.Event) error {
76 got = append(got, ev.GetId())
77 if ev.GetId() == "e3" {
78 cancel()
79 }
80 return nil
81 })
82 require.NoError(t, err, "stopping with the context is not an error")
83 assert.Equal(t, []string{"e1", "e2", "e3"}, got)
84
85 require.Len(t, requests, 4)
86 assert.Equal(t, "", requests[0].GetResumeAfter())
87 for _, r := range requests[1:3] {
88 assert.Equal(t, "e2", r.GetResumeAfter())
89 }
90 assert.Equal(t, "e2", requests[3].GetResumeAfter())
91 assert.Equal(t, []string{"home"}, requests[3].GetTopics())
92
93 // First wait: backoff was reset by the healthy first stream (Min), then it
94 // doubles and is capped at Max. Delays are jittered by ±20 %.
95 require.Len(t, sl.d, 3)
96 assertNear(t, time.Second, sl.d[0])
97 assertNear(t, 2*time.Second, sl.d[1])
98 assertNear(t, 3*time.Second, sl.d[2])
99
100 require.NotEmpty(t, notices)
101 assert.Contains(t, notices[0], "connection lost")
102 assert.Contains(t, notices[0], "shutting down")
103 assert.Contains(t, notices, "reconnected")
104}
105
106func assertNear(t *testing.T, want, got time.Duration) {
107 t.Helper()
108 assert.InDelta(t, float64(want), float64(got), float64(want)*0.21)
109}
110
111func TestWatchWarnsWhenThereIsNoResumePoint(t *testing.T) {
112 ctx, cancel := context.WithCancel(t.Context())
113 defer cancel()
114 calls := 0
115 f := &fake{}
116 f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
117 calls++
118 if calls == 2 {
119 cancel()
120 }
121 return connect.NewError(connect.CodeUnavailable, errors.New("down"))
122 }
123 c := start(t, f, client.Config{})
124 var notices []string
125 require.NoError(t, c.Watch(ctx, client.WatchOptions{
126 Sleep: (&sleeps{}).sleep,
127 OnNotice: func(m string) { notices = append(notices, m) },
128 }, func(*ermite.Event) error { return nil }))
129 require.NotEmpty(t, notices)
130 assert.Contains(t, notices[0], "no resume point")
131}
132
133func TestWatchStopsOnPermanentErrors(t *testing.T) {
134 for _, code := range []connect.Code{connect.CodeUnauthenticated, connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeInvalidArgument, connect.CodeUnimplemented} {
135 t.Run(code.String(), func(t *testing.T) {
136 calls := 0
137 f := &fake{}
138 f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
139 calls++
140 return connect.NewError(code, errors.New("nope"))
141 }
142 c := start(t, f, client.Config{})
143 err := c.Watch(t.Context(), client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(*ermite.Event) error { return nil })
144 require.Error(t, err)
145 assert.Equal(t, 1, calls, "no retry")
146 var ce *connect.Error
147 require.ErrorAs(t, err, &ce)
148 assert.Equal(t, code, ce.Code())
149 })
150 }
151}
152
153func TestWatchReturnsHandlerError(t *testing.T) {
154 f := &fake{}
155 f.watch = func(ctx context.Context, _ *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
156 _ = st.Send(pushed("e1"))
157 <-ctx.Done()
158 return nil
159 }
160 c := start(t, f, client.Config{})
161 boom := errors.New("write failed")
162 err := c.Watch(t.Context(), client.WatchOptions{}, func(*ermite.Event) error { return boom })
163 assert.ErrorIs(t, err, boom)
164}
165
166func TestWatchPassesGapAndIgnoresItAsResumePoint(t *testing.T) {
167 ctx, cancel := context.WithCancel(t.Context())
168 defer cancel()
169 var resumes []string
170 f := &fake{}
171 f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
172 resumes = append(resumes, req.Msg.GetResumeAfter())
173 if len(resumes) == 1 {
174 _ = st.Send(pushed("e1"))
175 _ = st.Send(&ermite.WatchResponse{Event: &ermite.Event{Kind: &ermite.Event_Gap{Gap: &ermite.Gap{}}}})
176 return connect.NewError(connect.CodeUnavailable, errors.New("x"))
177 }
178 _ = st.Send(pushed("e2"))
179 <-ctx.Done()
180 return nil
181 }
182 c := start(t, f, client.Config{})
183 var kinds []string
184 require.NoError(t, c.Watch(ctx, client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(ev *ermite.Event) error {
185 if ev.GetGap() != nil {
186 kinds = append(kinds, "gap")
187 } else {
188 kinds = append(kinds, ev.GetId())
189 }
190 if ev.GetId() == "e2" {
191 cancel()
192 }
193 return nil
194 }))
195 assert.Equal(t, []string{"e1", "gap", "e2"}, kinds)
196 assert.Equal(t, []string{"", "e1"}, resumes, "a gap has no id and must not erase the resume point")
197}
198
199func TestWatchResumesFromGivenID(t *testing.T) {
200 var first string
201 f := &fake{}
202 f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], _ *connect.ServerStream[ermite.WatchResponse]) error {
203 first = req.Msg.GetResumeAfter()
204 return connect.NewError(connect.CodePermissionDenied, errors.New("stop"))
205 }
206 c := start(t, f, client.Config{})
207 _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil })
208 assert.Equal(t, "abc", first)
209}
210
211// A connection cut mid-stream surfaces as a client-side invalid_argument; it
212// must be retried, unlike an invalid_argument the server sent.
213func TestWatchRetriesCutConnections(t *testing.T) {
214 ctx, cancel := context.WithCancel(t.Context())
215 defer cancel()
216 calls := 0
217 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
218 calls++
219 if calls == 1 {
220 // Announce a frame and hang up before sending it.
221 w.Header().Set("Content-Type", "application/connect+proto")
222 _, _ = w.Write([]byte{0, 0, 0, 0, 9})
223 return
224 }
225 cancel()
226 }))
227 defer srv.Close()
228 c, err := client.New(client.Config{URL: srv.URL})
229 require.NoError(t, err)
230 var notices []string
231 require.NoError(t, c.Watch(ctx, client.WatchOptions{
232 Sleep: (&sleeps{}).sleep,
233 OnNotice: func(m string) { notices = append(notices, m) },
234 }, func(*ermite.Event) error { return nil }))
235 assert.GreaterOrEqual(t, calls, 2, "retried")
236 require.NotEmpty(t, notices)
237 assert.Contains(t, notices[0], "connection lost")
238}