diff --git a/session.go b/session.go index 792727f..dae9f3f 100644 --- a/session.go +++ b/session.go @@ -19,6 +19,7 @@ import ( // ---------------------------------------------------------------------------- var ErrConnectionUnavailable = errors.New("connection unavailable") +var ErrReadLimit = transport.ErrReadLimit // ---------------------------------------------------------------------------- // Config @@ -249,6 +250,8 @@ func (w *session) serve( stopKeepalive, inactive, heartbeat := w.startKeepalive() // run session loop + var disconnectErr error + session: for { select { @@ -257,14 +260,13 @@ session: case data, ok := <-conn.Incoming(): if !ok { - var reason error select { - case reason = <-conn.Errors(): + case disconnectErr = <-conn.Errors(): default: - reason = fmt.Errorf("unknown") + disconnectErr = fmt.Errorf("unknown") } if w.logger != nil { - w.logger.Info("websocket: closed", "reason", reason) + w.logger.Info("websocket: closed", "reason", disconnectErr) } break session } @@ -297,7 +299,8 @@ session: // tear down connection w.conn.Store(nil) - pool.Events <- PoolEvent{URL: w.url, Kind: EventDisconnected, At: time.Now()} + pool.Events <- PoolEvent{ + URL: w.url, Kind: EventDisconnected, Err: disconnectErr, At: time.Now()} if w.logger != nil { w.logger.Info("disconnected") diff --git a/session_test.go b/session_test.go index 46bb086..cc57072 100644 --- a/session_test.go +++ b/session_test.go @@ -80,6 +80,37 @@ func TestSession(t *testing.T) { }, "expected EventConnected") }) + t.Run("EventDisconnected carries ErrReadLimit when read limit exceeded", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mockSocket := honeybeetest.NewMockSocket() + mockSocket.ReadMessageFunc = func() (int, []byte, error) { + return 0, nil, websocket.ErrReadLimit + } + + config, _ := NewSessionConfig( + WithRetryDisabled(), WithReconnectDelay(0)) + w := makeSession(t, mockSocket, config, ctx, cancel) + _, events, pool := makeSessionContext(t) + + var wg sync.WaitGroup + wg.Go(func() { + w.Start(pool) + }) + + honeybeetest.Eventually(t, func() bool { + select { + case e := <-events: + return e.Kind == EventDisconnected && + e.Err != nil && + errors.Is(e.Err, ErrReadLimit) + default: + return false + } + }, "expected EventDisconnected with Err matching ErrReadLimit") + }) + t.Run("dial failure exhausted - session exits cleanly, no connected/disconnected events", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/transport/connection.go b/transport/connection.go index 4f81da6..e3bd81d 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -27,6 +27,7 @@ var ( ErrPeerClosedClean = errors.New("peer closed connection cleanly") ErrPeerClosedUnexpected = errors.New("peer closed connection unexpectedly") ErrReadError = errors.New("read error") + ErrReadLimit = errors.New("read limit exceeded") ) // ---------------------------------------------------------------------------- @@ -350,7 +351,12 @@ func (c *Connection) classifyCloseError(err error) error { } } - classifiedError = fmt.Errorf("%w: %w", ErrReadError, err) + if errors.Is(err, websocket.ErrReadLimit) { + classifiedError = fmt.Errorf("%w: %w: %v", + ErrReadError, ErrReadLimit, err) + } else { + classifiedError = fmt.Errorf("%w: %w", ErrReadError, err) + } } return classifiedError diff --git a/transport/connection_test.go b/transport/connection_test.go index 404bf0e..0be187f 100644 --- a/transport/connection_test.go +++ b/transport/connection_test.go @@ -225,6 +225,51 @@ func TestConnection_Errors(t *testing.T) { }, "expected read error") }) + + t.Run("read limit exceeded", func(t *testing.T) { + mockSocket := honeybeetest.NewMockSocket() + mockSocket.ReadMessageFunc = func() (int, []byte, error) { + return 0, nil, websocket.ErrReadLimit + } + + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) + assert.NoError(t, err) + defer conn.Close() + + honeybeetest.Eventually(t, func() bool { + select { + case err := <-conn.Errors(): + return errors.Is(err, ErrReadLimit) && + errors.Is(err, ErrReadError) && + !errors.Is(err, websocket.ErrReadLimit) + default: + return false + } + }, "expected read limit error classified as ErrReadLimit") + + }) + + t.Run("generic read error is not a read limit error", func(t *testing.T) { + mockSocket := honeybeetest.NewMockSocket() + mockSocket.ReadMessageFunc = func() (int, []byte, error) { + return 0, nil, io.EOF + } + + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) + assert.NoError(t, err) + defer conn.Close() + + honeybeetest.Eventually(t, func() bool { + select { + case err := <-conn.Errors(): + return errors.Is(err, ErrReadError) && + !errors.Is(err, ErrReadLimit) + default: + return false + } + }, "expected generic read error without ErrReadLimit") + + }) } func TestConnection_Heartbeat(t *testing.T) {