Skip to content

Commit 8c8950b

Browse files
tstrombergtstromberg
committed
improve connectivity/context robustness
1 parent df14c80 commit 8c8950b

2 files changed

Lines changed: 86 additions & 88 deletions

File tree

‎pkg/client/client.go‎

Lines changed: 82 additions & 81 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"sync"
1313
"time"
1414

15+
"github.com/codeGROOVE-dev/retry"
1516
"golang.org/x/net/websocket"
1617
)
1718

@@ -36,6 +37,10 @@ const (
3637
// Server ping timing constants.
3738
expectedPingInterval = 54 * time.Second
3839
pingDelayThreshold = 10 * time.Second
40+
41+
// Read timeout for WebSocket operations.
42+
// Set to 90s to be longer than server ping interval (54s) to avoid false timeouts.
43+
readTimeout = 90 * time.Second
3944
)
4045

4146
// Event represents a webhook event received from the server.
@@ -118,33 +123,28 @@ func New(config Config) (*Client, error) {
118123
func (c *Client) Start(ctx context.Context) error {
119124
defer close(c.stoppedCh)
120125

121-
for {
122-
select {
123-
case <-ctx.Done():
124-
c.logger.Info("Client context cancelled, shutting down")
125-
return ctx.Err()
126-
case <-c.stopCh:
127-
c.logger.Info("Client stop requested")
128-
return nil
129-
default:
130-
}
126+
// Create retry options
127+
retryOpts := []retry.Option{
128+
retry.Context(ctx),
129+
retry.DelayType(retry.FullJitterBackoffDelay),
130+
retry.MaxDelay(c.config.MaxBackoff),
131+
retry.OnRetry(func(n uint, err error) {
132+
c.mu.Lock()
133+
//nolint:gosec // Retry count will not overflow in practice
134+
c.retries = int(n)
135+
c.mu.Unlock()
131136

132-
// Connection attempt logging
133-
if c.retries == 0 {
134-
c.logger.Info("========================================")
135-
c.logger.Info("CONNECTING to WebSocket server", "url", c.config.ServerURL)
136-
c.logger.Info("========================================")
137-
} else {
138-
c.logger.Info("========================================")
139-
c.logger.Info("RECONNECTING to WebSocket server", "url", c.config.ServerURL, "attempt", c.retries)
140-
c.logger.Info("========================================")
141-
}
137+
c.logger.Warn(separatorLine)
138+
c.logger.Warn("WebSocket CONNECTION LOST!", "error", err, "events_received", c.eventCount, "attempt", n+1)
139+
c.logger.Warn(separatorLine)
142140

143-
// Try to connect
144-
err := c.connect(ctx)
145-
// Handle connection result
146-
if err != nil {
147-
// Check if it's an authentication error - don't retry these
141+
// Notify disconnect callback
142+
if c.config.OnDisconnect != nil {
143+
c.config.OnDisconnect(err)
144+
}
145+
}),
146+
retry.RetryIf(func(err error) bool {
147+
// Don't retry authentication errors
148148
var authErr *AuthenticationError
149149
if errors.As(err, &authErr) {
150150
c.logger.Error(separatorLine)
@@ -154,51 +154,63 @@ func (c *Client) Start(ctx context.Context) error {
154154
c.logger.Error("- Not being a member of the requested organization")
155155
c.logger.Error("- Insufficient permissions")
156156
c.logger.Error(separatorLine)
157-
return err
158-
}
159-
160-
c.logger.Warn(separatorLine)
161-
c.logger.Warn("WebSocket CONNECTION LOST!", "error", err, "events_received", c.eventCount)
162-
c.logger.Warn(separatorLine)
163-
164-
// Notify disconnect callback
165-
if c.config.OnDisconnect != nil {
166-
c.config.OnDisconnect(err)
157+
return false
167158
}
168159

169-
// Check if reconnection is disabled
160+
// Don't retry if reconnection is disabled
170161
if c.config.NoReconnect {
171-
return fmt.Errorf("connection failed and reconnection disabled: %w", err)
162+
return false
172163
}
173164

174-
// Check retry limit
175-
c.retries++
176-
if c.config.MaxRetries > 0 && c.retries > c.config.MaxRetries {
177-
c.logger.Error("Exceeded maximum retry attempts. Giving up.", "max_retries", c.config.MaxRetries)
178-
return fmt.Errorf("exceeded maximum retry attempts (%d)", c.config.MaxRetries)
165+
// Don't retry if stop was requested
166+
select {
167+
case <-c.stopCh:
168+
return false
169+
default:
170+
return true
179171
}
172+
}),
173+
}
180174

181-
// Calculate backoff delay
182-
delay := time.Duration(c.retries) * time.Second
183-
if delay > c.config.MaxBackoff {
184-
delay = c.config.MaxBackoff
185-
}
175+
// Configure retry attempts
176+
if c.config.MaxRetries > 0 {
177+
//nolint:gosec // MaxRetries is a user-configured value, overflow not a concern
178+
retryOpts = append(retryOpts, retry.Attempts(uint(c.config.MaxRetries)))
179+
} else {
180+
retryOpts = append(retryOpts, retry.UntilSucceeded())
181+
}
186182

187-
c.logger.Info("Will attempt to reconnect", "delay_seconds", delay.Seconds())
188-
c.logger.Info("Press Ctrl+C to exit")
183+
// Use retry library to handle reconnection with exponential backoff and jitter
184+
return retry.Do(func() error {
185+
// Check for early cancellation - don't retry on shutdown
186+
select {
187+
case <-ctx.Done():
188+
c.logger.Info("Client context cancelled, shutting down")
189+
return retry.Unrecoverable(ctx.Err())
190+
case <-c.stopCh:
191+
c.logger.Info("Client stop requested")
192+
return retry.Unrecoverable(errors.New("stop requested"))
193+
default:
194+
}
189195

190-
// Wait before reconnecting
191-
select {
192-
case <-time.After(delay):
193-
c.logger.Info("Reconnection delay elapsed, attempting to reconnect")
194-
continue
195-
case <-ctx.Done():
196-
return ctx.Err()
197-
case <-c.stopCh:
198-
return nil
199-
}
196+
// Connection attempt logging
197+
c.mu.RLock()
198+
n := c.retries
199+
c.mu.RUnlock()
200+
201+
if n == 0 {
202+
c.logger.Info("========================================")
203+
c.logger.Info("CONNECTING to WebSocket server", "url", c.config.ServerURL)
204+
c.logger.Info("========================================")
205+
} else {
206+
c.logger.Info("========================================")
207+
c.logger.Info("RECONNECTING to WebSocket server", "url", c.config.ServerURL, "attempt", n)
208+
c.logger.Info("========================================")
200209
}
201-
}
210+
211+
// Try to connect - this will run indefinitely if successful
212+
return c.connect(ctx)
213+
}, retryOpts...)
202214
}
203215

204216
// Stop gracefully stops the client.
@@ -242,7 +254,10 @@ func (c *Client) connect(ctx context.Context) error {
242254
// Extract status code if present
243255
if strings.Contains(errStr, "403") || strings.Contains(errLower, "forbidden") {
244256
return &AuthenticationError{
245-
message: fmt.Sprintf("Authentication failed (403 Forbidden): Check your GitHub token and organization membership. Original error: %v", err),
257+
message: fmt.Sprintf(
258+
"Authentication failed (403 Forbidden): Check your GitHub token and organization membership. Original error: %v",
259+
err,
260+
),
246261
}
247262
}
248263
if strings.Contains(errStr, "401") || strings.Contains(errLower, "unauthorized") {
@@ -320,21 +335,12 @@ func (c *Client) connect(ctx context.Context) error {
320335
}
321336

322337
// Check response type
323-
responseType := ""
324-
if t, ok := firstResponse[msgTypeField].(string); ok {
325-
responseType = t
326-
}
338+
responseType, _ := firstResponse[msgTypeField].(string) //nolint:errcheck // type assertion, not error
327339

328340
// Handle error response
329341
if responseType == "error" {
330-
errorCode := ""
331-
if code, ok := firstResponse["error"].(string); ok {
332-
errorCode = code
333-
}
334-
message := ""
335-
if msg, ok := firstResponse["message"].(string); ok {
336-
message = msg
337-
}
342+
errorCode, _ := firstResponse["error"].(string) //nolint:errcheck // type assertion, not error
343+
message, _ := firstResponse["message"].(string) //nolint:errcheck // type assertion, not error
338344
c.logger.Error(separatorLine)
339345
c.logger.Error("SUBSCRIPTION REJECTED BY SERVER!", "error_code", errorCode, "message", message)
340346
c.logger.Error(separatorLine)
@@ -384,7 +390,9 @@ func (c *Client) connect(ctx context.Context) error {
384390
}
385391

386392
// Reset retry counter on successful connection
393+
c.mu.Lock()
387394
c.retries = 0
395+
c.mu.Unlock()
388396

389397
// Start ping sender
390398
pingCtx, cancelPing := context.WithCancel(ctx)
@@ -423,10 +431,6 @@ func (c *Client) sendPings(ctx context.Context, ws *websocket.Conn) {
423431

424432
// readEvents reads and processes events from the WebSocket with responsive shutdown.
425433
func (c *Client) readEvents(ctx context.Context, ws *websocket.Conn) error {
426-
// Set read timeout to be longer than server ping interval (54s) to avoid false timeouts
427-
// Server sends pings every 54s, so we should expect a message at least that often
428-
readTimeout := 90 * time.Second
429-
430434
for {
431435
// Check for context cancellation first
432436
select {
@@ -470,10 +474,7 @@ func (c *Client) readEvents(ctx context.Context, ws *websocket.Conn) error {
470474
}
471475

472476
// Check message type
473-
responseType := ""
474-
if t, ok := response[msgTypeField].(string); ok {
475-
responseType = t
476-
}
477+
responseType, _ := response[msgTypeField].(string) //nolint:errcheck // type assertion, not error
477478

478479
// Handle ping messages
479480
if responseType == "ping" {

‎pkg/github/client.go‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ func (c *Client) AuthenticatedUser(ctx context.Context) (*User, error) {
5656
var user *User
5757
var lastErr error
5858

59-
// Retry with exponential backoff and jitter for transient failures
59+
// Retry with exponential backoff and full jitter for transient failures
6060
err := retry.Do(
6161
func() error {
6262
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://api.github.com/user", http.NoBody)
@@ -134,9 +134,8 @@ func (c *Client) AuthenticatedUser(ctx context.Context) (*User, error) {
134134
}
135135
},
136136
retry.Attempts(3),
137-
retry.DelayType(retry.BackOffDelay),
137+
retry.DelayType(retry.FullJitterBackoffDelay),
138138
retry.MaxDelay(2*time.Minute),
139-
retry.MaxJitter(time.Second),
140139
retry.Context(ctx),
141140
)
142141
if err != nil {
@@ -250,9 +249,8 @@ func (c *Client) AppInstallationInfo(ctx context.Context) (*AppInstallation, err
250249
}
251250
},
252251
retry.Attempts(3),
253-
retry.DelayType(retry.BackOffDelay),
252+
retry.DelayType(retry.FullJitterBackoffDelay),
254253
retry.MaxDelay(2*time.Minute),
255-
retry.MaxJitter(time.Second),
256254
retry.Context(ctx),
257255
)
258256
if err != nil {
@@ -413,9 +411,8 @@ func (c *Client) userOrganizations(ctx context.Context) ([]Organization, error)
413411
}
414412
},
415413
retry.Attempts(3),
416-
retry.DelayType(retry.BackOffDelay),
414+
retry.DelayType(retry.FullJitterBackoffDelay),
417415
retry.MaxDelay(2*time.Minute),
418-
retry.MaxJitter(time.Second),
419416
retry.Context(ctx),
420417
)
421418
if err != nil {

0 commit comments

Comments
 (0)