diff --git a/go.mod b/go.mod index 90d5344..869bc30 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module nudgebee/forager -go 1.26.9 +go 1.26.10 require ( cloud.google.com/go/auth v0.23.1 diff --git a/pkg/ws/client.go b/pkg/ws/client.go index 6865fe7..66dcaad 100644 --- a/pkg/ws/client.go +++ b/pkg/ws/client.go @@ -4,11 +4,14 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "log/slog" + "net" "net/http" "sync" + "syscall" "time" "github.com/gorilla/websocket" @@ -119,7 +122,11 @@ func (c *Client) Run(ctx context.Context) error { err := c.connectAndServe(ctx) if err != nil { - c.logger.Error("websocket session ended", "err", err) + if isRoutineDisconnect(err) { + c.logger.Warn("websocket session ended", "err", err) + } else { + c.logger.Error("websocket session ended", "err", err) + } } select { @@ -153,7 +160,11 @@ func (c *Client) connectAndServe(ctx context.Context) error { body, _ = io.ReadAll(io.LimitReader(resp.Body, 1024)) resp.Body.Close() // nolint:errcheck } - return fmt.Errorf("dial failed: %w (status=%d body=%q)", err, resp.StatusCode, body) + return &dialHTTPError{ + StatusCode: resp.StatusCode, + Body: body, + Err: err, + } } return fmt.Errorf("dial failed: %w", err) } @@ -440,3 +451,56 @@ func (c *Client) Send(msg []byte) { c.logger.Warn("send channel full, dropping message") } } + +type dialHTTPError struct { + StatusCode int + Body []byte + Err error +} + +func (e *dialHTTPError) Error() string { + return fmt.Sprintf("dial failed: %v (status=%d body=%q)", e.Err, e.StatusCode, e.Body) +} + +func (e *dialHTTPError) Unwrap() error { + return e.Err +} + +// isRoutineDisconnect returns true if err is a routine network drop, timeout, +// normal/abnormal websocket closure, or temporary dial failure. +func isRoutineDisconnect(err error) bool { + if err == nil { + return false + } + + var httpErr *dialHTTPError + if errors.As(err, &httpErr) { + // 4xx status codes (such as 401 Unauthorized or 403 Forbidden) are + // persistent client/auth configuration errors. + if httpErr.StatusCode >= 400 && httpErr.StatusCode < 500 { + return false + } + // 5xx status codes are temporary upstream/relay server issues. + return true + } + + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return true + } + + var closeErr *websocket.CloseError + if errors.As(err, &closeErr) { + return true + } + + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return true + } + + if errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.EPIPE) { + return true + } + + var netErr net.Error + return errors.As(err, &netErr) +} diff --git a/pkg/ws/client_disconnect_test.go b/pkg/ws/client_disconnect_test.go new file mode 100644 index 0000000..a1ffdb9 --- /dev/null +++ b/pkg/ws/client_disconnect_test.go @@ -0,0 +1,159 @@ +package ws + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "syscall" + "testing" + + "github.com/gorilla/websocket" +) + +type customNetError struct{} + +func (e *customNetError) Error() string { return "custom net error" } +func (e *customNetError) Timeout() bool { return true } +func (e *customNetError) Temporary() bool { return true } + +func TestIsRoutineDisconnect(t *testing.T) { + tests := []struct { + name string + err error + expected bool + }{ + { + name: "nil error", + err: nil, + expected: false, + }, + { + name: "dialHTTPError 401 unauthorized", + err: &dialHTTPError{ + StatusCode: 401, + Body: []byte("unauthorized"), + Err: errors.New("bad status"), + }, + expected: false, + }, + { + name: "dialHTTPError 403 forbidden", + err: &dialHTTPError{ + StatusCode: 403, + Body: []byte("forbidden"), + Err: errors.New("bad status"), + }, + expected: false, + }, + { + name: "dialHTTPError 500 internal server error", + err: &dialHTTPError{ + StatusCode: 500, + Body: []byte("internal error"), + Err: errors.New("bad status"), + }, + expected: true, + }, + { + name: "dialHTTPError 502 bad gateway", + err: &dialHTTPError{ + StatusCode: 502, + Body: []byte("bad gateway"), + Err: errors.New("bad status"), + }, + expected: true, + }, + { + name: "context canceled", + err: context.Canceled, + expected: true, + }, + { + name: "context deadline exceeded", + err: context.DeadlineExceeded, + expected: true, + }, + { + name: "websocket close error", + err: &websocket.CloseError{ + Code: websocket.CloseGoingAway, + Text: "going away", + }, + expected: true, + }, + { + name: "io EOF", + err: io.EOF, + expected: true, + }, + { + name: "io ErrUnexpectedEOF", + err: io.ErrUnexpectedEOF, + expected: true, + }, + { + name: "syscall ECONNRESET", + err: syscall.ECONNRESET, + expected: true, + }, + { + name: "syscall ECONNREFUSED", + err: syscall.ECONNREFUSED, + expected: true, + }, + { + name: "syscall EPIPE", + err: syscall.EPIPE, + expected: true, + }, + { + name: "net.Error", + err: &customNetError{}, + expected: true, + }, + { + name: "wrapped net.OpError", + err: &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("timeout")}, + expected: true, + }, + { + name: "wrapped routine error", + err: fmt.Errorf("connection failed: %w", syscall.ECONNRESET), + expected: true, + }, + { + name: "arbitrary internal error", + err: errors.New("unhandled failure"), + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isRoutineDisconnect(tt.err) + if got != tt.expected { + t.Errorf("isRoutineDisconnect(%v) = %v, want %v", tt.err, got, tt.expected) + } + }) + } +} + +func TestDialHTTPError(t *testing.T) { + underlying := errors.New("handshake failed") + err := &dialHTTPError{ + StatusCode: 401, + Body: []byte("unauthorized access"), + Err: underlying, + } + + if !errors.Is(err, underlying) { + t.Errorf("errors.Is(err, underlying) = false, want true") + } + + expectedStr := `dial failed: handshake failed (status=401 body="unauthorized access")` + if err.Error() != expectedStr { + t.Errorf("err.Error() = %q, want %q", err.Error(), expectedStr) + } +}