Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
module nudgebee/forager

go 1.26.9
go 1.26.10

require (
cloud.google.com/go/auth v0.23.1
Expand Down
68 changes: 66 additions & 2 deletions pkg/ws/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
159 changes: 159 additions & 0 deletions pkg/ws/client_disconnect_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
Loading