1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package client
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "math/rand/v2"
10 "time"
11
12 "connectrpc.com/connect"
13 "github.com/bnema/zerowrap"
14
15 "bnema.dev/ermite-api/gen/ermite"
16)
17
18// Backoff is the delay between reconnections: it starts at Min, doubles after
19// each failed attempt up to Max and resets once a connection has proved
20// healthy. Delays are jittered by ±20 %.
21type Backoff struct {
22 Min time.Duration
23 Max time.Duration
24}
25
26// DefaultBackoff is 500 ms growing to 30 s.
27var DefaultBackoff = Backoff{Min: 500 * time.Millisecond, Max: 30 * time.Second}
28
29// healthyAfter is how long a stream must stay up to count as healthy when it
30// delivered no event, for example on a quiet server.
31const healthyAfter = 10 * time.Second
32
33// WatchOptions configures Watch.
34type WatchOptions struct {
35 // Topics are the subjects to follow, see WatchRequest.topics.
36 Topics []string
37 // ResumeAfter is the id of the last event already seen, or "".
38 ResumeAfter string
39 // Backoff overrides DefaultBackoff when Min is set.
40 Backoff Backoff
41 // Sleep waits for d or until ctx ends; nil sleeps for real. For tests.
42 Sleep func(ctx context.Context, d time.Duration) error
43 // OnNotice receives connection notices (lost, retrying, back); optional.
44 OnNotice func(msg string)
45}
46
47// Watch follows EventService.Watch and calls onEvent for each event, in
48// order, including Gap events (the caller decides how to tell the user). When
49// the connection drops or the server closes the stream it reconnects with
50// backoff, resuming after the last event id it saw. It returns nil when ctx
51// ends, onEvent's error if it fails, and a described error when the server
52// refuses the call for a reason a retry cannot fix (authentication, access,
53// invalid topic, unsupported call).
54func (c *Client) Watch(ctx context.Context, opts WatchOptions, onEvent func(*ermite.Event) error) error {
55 log := zerowrap.FromCtx(ctx)
56 bo := opts.Backoff
57 if bo.Min <= 0 {
58 bo = DefaultBackoff
59 }
60 bo.Max = max(bo.Max, bo.Min)
61 sleep := opts.Sleep
62 if sleep == nil {
63 sleep = sleepCtx
64 }
65 notice := func(msg string) {
66 if opts.OnNotice != nil {
67 opts.OnNotice(msg)
68 }
69 }
70
71 events := c.Events()
72 last := opts.ResumeAfter
73 delay := bo.Min
74 retrying := false
75 for {
76 started := time.Now()
77 received := false
78 stream, err := events.Watch(ctx, connect.NewRequest(&ermite.WatchRequest{Topics: opts.Topics, ResumeAfter: last}))
79 if err == nil {
80 for stream.Receive() {
81 if !received && retrying {
82 notice("reconnected")
83 }
84 received = true
85 ev := stream.Msg().GetEvent()
86 if ev == nil {
87 continue
88 }
89 if err := onEvent(ev); err != nil {
90 _ = stream.Close()
91 return err
92 }
93 if ev.GetId() != "" {
94 last = ev.GetId()
95 }
96 }
97 err = stream.Err()
98 _ = stream.Close()
99 }
100 if ctx.Err() != nil {
101 return nil
102 }
103 if err != nil && !retryable(err) {
104 return c.Describe(err)
105 }
106
107 if received || time.Since(started) >= healthyAfter {
108 delay = bo.Min
109 }
110 wait := jitter(delay)
111 reason := "the server closed the stream"
112 if err != nil {
113 reason = c.Describe(err).Error()
114 }
115 msg := fmt.Sprintf("connection lost (%s); reconnecting in %s", reason, wait.Round(10*time.Millisecond))
116 if last == "" {
117 msg += " (no resume point yet: events during the outage will be missed)"
118 }
119 log.Debug().Str("reason", reason).Dur("wait", wait).Str("resume_after", last).Msg("watch reconnecting")
120 notice(msg)
121 if err := sleep(ctx, wait); err != nil {
122 return nil
123 }
124 delay = min(delay*2, bo.Max)
125 retrying = true
126 }
127}
128
129// retryable reports whether a reconnection can fix the error. The caller has
130// checked that its own context is alive, so a cancellation here comes from the
131// network (a proxy resetting the stream).
132func retryable(err error) bool {
133 var ce *connect.Error
134 if !errors.As(err, &ce) {
135 return true
136 }
137 // A connection cut in the middle of a stream is reported by Connect as an
138 // invalid_argument "incomplete envelope" error that the server never sent.
139 // Only an invalid_argument that came over the wire (a bad topic) is final.
140 if ce.Code() == connect.CodeInvalidArgument && !connect.IsWireError(err) {
141 return true
142 }
143 switch ce.Code() {
144 case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded,
145 connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled:
146 return true
147 default:
148 return false
149 }
150}
151
152func jitter(d time.Duration) time.Duration {
153 return time.Duration(float64(d) * (0.8 + 0.4*rand.Float64()))
154}
155
156func sleepCtx(ctx context.Context, d time.Duration) error {
157 t := time.NewTimer(d)
158 defer t.Stop()
159 select {
160 case <-t.C:
161 return nil
162 case <-ctx.Done():
163 return ctx.Err()
164 }
165}