diff --git a/pkg/listen/proxy/proxy.go b/pkg/listen/proxy/proxy.go index 6d8367e2..fd16a656 100644 --- a/pkg/listen/proxy/proxy.go +++ b/pkg/listen/proxy/proxy.go @@ -14,6 +14,7 @@ import ( "os/signal" "strconv" "strings" + "sync" "sync/atomic" "syscall" "time" @@ -50,7 +51,7 @@ type Config struct { // Disable periodic health checks of the local server NoHealthcheck bool // Output mode: interactive, compact, quiet - Output string + Output string GuestURL string // MaxConnections allows tuning the maximum concurrent connections per host. // Default: 50 concurrent connections @@ -69,15 +70,19 @@ type Config struct { // webhook events, forwards them to the local endpoint and sends the response // back to Hookdeck. type Proxy struct { - cfg *Config - connections []*hookdeck.Connection - webSocketClient *websocket.Client - connectionTimer *time.Timer - httpClient *http.Client - transport *http.Transport - activeRequests int32 - maxConnWarned bool // Track if we've warned about connection limit - renderer Renderer + cfg *Config + connections []*hookdeck.Connection + // webSocketClient is reassigned by Run's reconnect loop and read from other goroutines + // (signal handler, event handlers). Always access it through currentWebSocketClient / + // setWebSocketClient. + webSocketClient *websocket.Client + webSocketClientMu sync.Mutex + connectionTimer *time.Timer + httpClient *http.Client + transport *http.Transport + activeRequests int32 + maxConnWarned bool // Track if we've warned about connection limit + renderer Renderer // Server health monitoring serverHealthy atomic.Bool @@ -97,6 +102,20 @@ func withSIGTERMCancel(ctx context.Context, onCancel func()) context.Context { return ctx } +// currentWebSocketClient returns the active websocket client (nil before the first connect). +// Guards against the reconnect loop reassigning the client while another goroutine reads it. +func (p *Proxy) currentWebSocketClient() *websocket.Client { + p.webSocketClientMu.Lock() + defer p.webSocketClientMu.Unlock() + return p.webSocketClient +} + +func (p *Proxy) setWebSocketClient(client *websocket.Client) { + p.webSocketClientMu.Lock() + defer p.webSocketClientMu.Unlock() + p.webSocketClient = client +} + // Run manages the connection to Hookdeck. // The connection is established in phases: // - Create a new CLI session @@ -110,19 +129,38 @@ func (p *Proxy) Run(parentCtx context.Context) error { // of connection attempts that will be made and will retry // until the connection is successful or the user terminates // the program. - hasConnectedOnce := false + // Atomic: written by the per-attempt connection monitor goroutine below and + // read by canConnect on this goroutine. + var hasConnectedOnce atomic.Bool canConnect := func() bool { - if hasConnectedOnce { + if hasConnectedOnce.Load() { return true } else { return nAttempts < maxConnectAttempts } } + // Set before the websocket is stopped below, so the reconnect loop can tell an + // intentional shutdown from a real disconnect. Stopping the client closes its + // NotifyExpired channel, which would otherwise race the context cancellation and + // make the loop announce "Connection lost, reconnecting..." as the CLI exits. + var shuttingDown atomic.Bool + signalCtx := withSIGTERMCancel(parentCtx, func() { log.WithFields(log.Fields{ "prefix": "proxy.Proxy.Run", }).Debug("Ctrl+C received, cleaning up...") + + shuttingDown.Store(true) + + // Send a clean WebSocket close (1000) before the context is + // cancelled. This lets the server tombstone the session + // immediately instead of holding it for the 2-minute grace + // window, so subsequent events don't get routed to a + // disconnected CLI and old sessions don't pile up. + if wsClient := p.currentWebSocketClient(); wsClient != nil { + wsClient.Stop() + } }) // Notify renderer we're connecting @@ -141,33 +179,49 @@ func (p *Proxy) Run(parentCtx context.Context) error { return fmt.Errorf("error while starting a new session") } + // Build session data to send on every connect/reconnect so the server + // can recreate the session if it expired in Redis. + var connectionIDs []string + for _, connection := range p.connections { + connectionIDs = append(connectionIDs, connection.ID) + } + + var filtersJSON string + if p.cfg.Filters != nil { + if b, err := json.Marshal(p.cfg.Filters); err == nil { + filtersJSON = string(b) + } + } + // Main loop to keep attempting to connect to Hookdeck once // we have created a session. for canConnect() { - p.webSocketClient = websocket.NewClient( + wsClient := websocket.NewClient( p.cfg.WSBaseURL, session.Id, p.cfg.Key, p.cfg.ProjectID, + connectionIDs, + filtersJSON, &websocket.Config{ Log: p.cfg.Log, NoWSS: p.cfg.NoWSS, EventHandler: websocket.EventHandlerFunc(p.processAttempt), }, ) + p.setWebSocketClient(wsClient) // Monitor the websocket for connection go func() { - <-p.webSocketClient.Connected() + <-wsClient.Connected() p.renderer.OnConnected() // Only start health monitoring on first successful connection to prevent - // goroutine leaks on reconnects. The hasConnectedOnce guard ensures that - // even if the websocket reconnects multiple times (which happens in the - // Run() loop), we only spawn the health monitor goroutine once. - if !hasConnectedOnce { - hasConnectedOnce = true - + // goroutine leaks on reconnects. The compare-and-swap ensures that even + // if the websocket reconnects multiple times (which happens in the Run() + // loop, each attempt spawning its own monitor goroutine), we only spawn + // the health monitor goroutine once. + if hasConnectedOnce.CompareAndSwap(false, true) { // Skip health monitoring if disabled via --no-healthcheck flag if p.cfg.NoHealthcheck { // Assume server is healthy when healthchecks are disabled @@ -185,22 +239,38 @@ func (p *Proxy) Run(parentCtx context.Context) error { }() // Run the websocket in the background - go p.webSocketClient.Run(signalCtx) + go wsClient.Run(signalCtx) nAttempts++ // Block until ctrl+c, renderer quit, or websocket connection is interrupted select { case <-signalCtx.Done(): + // The clean close (Stop) already ran in the withSIGTERMCancel callback, + // before the context was cancelled. return nil case <-p.renderer.Done(): // Renderer wants to quit (user pressed q or similar) - if p.webSocketClient != nil { - p.webSocketClient.Stop() - } + wsClient.Stop() p.renderer.Cleanup() return nil - case <-p.webSocketClient.NotifyExpired: + case <-wsClient.NotifyExpired: + // Stopping the client on shutdown closes NotifyExpired, so this case can + // win the race against signalCtx.Done(). That's an intentional exit, not a + // dropped connection — don't tell the user we're reconnecting. + if shuttingDown.Load() { + p.renderer.Cleanup() + return nil + } + p.renderer.OnDisconnected() + // If this attempt connected successfully before dropping (e.g. a + // routine server deploy closing with 1001), reset the counter so + // backoff reflects consecutive failures, not lifetime reconnects. + // Without this, a long-running CLI drifts toward the maximum + // backoff even though every reconnect succeeds immediately. + if wsClient.HasConnected() { + nAttempts = 0 + } if !canConnect() { p.renderer.Cleanup() return fmt.Errorf("Could not connect. Terminating after %d failed attempts to establish a connection.", nAttempts) @@ -240,8 +310,8 @@ func (p *Proxy) Run(parentCtx context.Context) error { } } - if p.webSocketClient != nil { - p.webSocketClient.Stop() + if wsClient := p.currentWebSocketClient(); wsClient != nil { + wsClient.Stop() } // Clean up renderer @@ -378,7 +448,7 @@ func (p *Proxy) processAttempt(msg websocket.IncomingMessage) { timer.Stop() if result.err != nil { p.renderer.OnEventError(eventID, webhookEvent, result.err, requestStartTime) - p.webSocketClient.SendMessage(&websocket.OutgoingMessage{ + p.currentWebSocketClient().SendMessage(&websocket.OutgoingMessage{ ErrorAttemptResponse: &websocket.ErrorAttemptResponse{ Event: "attempt_response", Body: websocket.ErrorAttemptBody{ @@ -405,7 +475,7 @@ func (p *Proxy) processAttempt(msg websocket.IncomingMessage) { if eventShown { if result.err != nil { p.renderer.OnEventError(eventID, webhookEvent, result.err, requestStartTime) - p.webSocketClient.SendMessage(&websocket.OutgoingMessage{ + p.currentWebSocketClient().SendMessage(&websocket.OutgoingMessage{ ErrorAttemptResponse: &websocket.ErrorAttemptResponse{ Event: "attempt_response", Body: websocket.ErrorAttemptBody{ @@ -445,8 +515,8 @@ func (p *Proxy) processEndpointResponse(eventID string, webhookEvent *websocket. }, requestStartTime) // Send response back to Hookdeck - if p.webSocketClient != nil { - p.webSocketClient.SendMessage(&websocket.OutgoingMessage{ + if wsClient := p.currentWebSocketClient(); wsClient != nil { + wsClient.SendMessage(&websocket.OutgoingMessage{ AttemptResponse: &websocket.AttemptResponse{ Event: "attempt_response", Body: websocket.AttemptResponseBody{ diff --git a/pkg/websocket/client.go b/pkg/websocket/client.go index 675a7dff..505e1611 100644 --- a/pkg/websocket/client.go +++ b/pkg/websocket/client.go @@ -69,9 +69,23 @@ type Client struct { TeamID string - // ID sent by the client in the `Websocket-Id` header when connecting + // WebSocketID is the CLI session ID (e.g., "cses_DPlA9BeXxNT2rT"). + // Sent as the `Websocket-Id` header. The server uses this to look up + // the session in Redis. This is NOT the same as ConnectionIDs below. WebSocketID string + // ConnectionIDs are the webhook/connection IDs (e.g., ["web_abc", "web_def"]) + // that this CLI session is listening on. Sent as the `X-Webhook-Ids` header + // on every connect/reconnect so the server can recreate the session in Redis + // if it expired. These map to `webhook_ids` on the session and are used for + // routing events to this CLI. + ConnectionIDs []string + + // FiltersJSON is the JSON-encoded session filters (e.g., '{"body":{"action":"opened"}}'). + // Sent base64-encoded as the `X-Session-Filters` header on every connect/reconnect. + // Empty string means no filters. + FiltersJSON string + // Feature that the websocket is specified for //WebSocketAuthorizedFeature string @@ -80,7 +94,11 @@ type Client struct { conn *ws.Conn done chan struct{} + doneOnce sync.Once isConnected bool + // stateMu guards conn and isConnected: they are written by the connect goroutine and + // read by Stop(), which can run on the signal-handler goroutine. + stateMu sync.Mutex NotifyExpired chan struct{} notifyClose chan error @@ -96,7 +114,7 @@ func (c *Client) Connected() <-chan struct{} { d := make(chan struct{}) go func() { - for !c.isConnected { + for !c.connected() { time.Sleep(100 * time.Millisecond) } close(d) @@ -105,9 +123,28 @@ func (c *Client) Connected() <-chan struct{} { return d } +func (c *Client) connected() bool { + c.stateMu.Lock() + defer c.stateMu.Unlock() + return c.isConnected +} + +// HasConnected reports whether this client successfully established its +// websocket connection at some point. It stays true after a disconnect, so +// callers can distinguish "connected then dropped" from "never connected". +func (c *Client) HasConnected() bool { + return c.connected() +} + +func (c *Client) setConnected(isConnected bool) { + c.stateMu.Lock() + c.isConnected = isConnected + c.stateMu.Unlock() +} + // Run starts listening for incoming webhook requests from Hookdeck. func (c *Client) Run(ctx context.Context) { - c.isConnected = false + c.setConnected(false) c.cfg.Log.WithFields(log.Fields{ "prefix": "websocket.client.Run", }).Debug("Attempting to connect to Hookdeck") @@ -162,9 +199,37 @@ func (c *Client) ConnectionLost() { c.NotifyExpired <- struct{}{} } -// Stop stops listening for incoming webhook events. +// Stop stops listening for incoming webhook events. It is safe to call +// multiple times. When called while connected, it sends a clean WebSocket +// close (code 1000) so the server can distinguish an intentional shutdown +// from an abnormal disconnect (network drop, crash). The server treats a +// 1000 close as a final tombstone and removes the session immediately +// instead of holding it open for the reconnect grace window. func (c *Client) Stop() { - close(c.done) + c.doneOnce.Do(func() { + // Snapshot the connection under stateMu — Stop can run on the + // signal-handler goroutine while the connect goroutine writes it. + // conn alone signals an established connection: it is only assigned + // after a successful upgrade, and checking isConnected too would skip + // the clean close in the window between changeConnection() and + // setConnected(true). + c.stateMu.Lock() + conn := c.conn + c.stateMu.Unlock() + + // If we have an active connection, send a clean close frame BEFORE + // tearing down the pumps. This guarantees the server sees code 1000 + // rather than the abnormal 1006 it gets when the TCP socket dies. + if conn != nil { + deadline := time.Now().Add(c.cfg.WriteWait) + _ = conn.WriteControl( + ws.CloseMessage, + ws.FormatCloseMessage(ws.CloseNormalClosure, "client_shutdown"), + deadline, + ) + } + close(c.done) + }) } // SendMessage sends a message to Hookdeck through the websocket. @@ -222,6 +287,17 @@ func (c *Client) connect(ctx context.Context) error { header.Set("X-Team-Id", c.TeamID) header.Set("Authorization", "Basic "+basicAuth(c.CLIKey, "")) + // Send session data on every connect/reconnect so the server can + // recreate the session if it expired in Redis between reconnects. + // Filters are base64-encoded: raw UTF-8 header bytes would be decoded as + // latin-1 by the Node server and silently corrupt non-ASCII filter values. + if len(c.ConnectionIDs) > 0 { + header.Set("X-Webhook-Ids", strings.Join(c.ConnectionIDs, ",")) + } + if c.FiltersJSON != "" { + header.Set("X-Session-Filters", base64.StdEncoding.EncodeToString([]byte(c.FiltersJSON))) + } + url := c.URL if c.cfg.NoWSS && strings.HasPrefix(url, "wss") { url = "ws" + strings.TrimPrefix(c.URL, "wss") @@ -249,7 +325,7 @@ func (c *Client) connect(ctx context.Context) error { defer resp.Body.Close() c.changeConnection(conn) - c.isConnected = true + c.setConnected(true) c.wg = &sync.WaitGroup{} c.wg.Add(2) @@ -267,7 +343,9 @@ func (c *Client) connect(ctx context.Context) error { // changeConnection takes a new connection and recreates the channels. func (c *Client) changeConnection(conn *ws.Conn) { + c.stateMu.Lock() c.conn = conn + c.stateMu.Unlock() c.notifyClose = make(chan error) c.stopReadPump = make(chan struct{}) c.stopWritePump = make(chan struct{}) @@ -309,22 +387,44 @@ func (c *Client) readPump() { "prefix": "websocket.Client.readPump", }).Debug("stopReadPump") default: + var closeErr *ws.CloseError switch { - case !ws.IsCloseError(err): + case !errors.As(err, &closeErr): // read errors do not prevent websocket reconnects in the CLI so we should // only display this on debug-level logging c.cfg.Log.WithFields(log.Fields{ "prefix": "websocket.Client.Close", }).Debug("read error: ", err) - case ws.IsUnexpectedCloseError(err, ws.CloseNormalClosure): + case closeErr.Code == ws.CloseNormalClosure: c.cfg.Log.WithFields(log.Fields{ "prefix": "websocket.Client.Close", - }).Error("close error: ", err) + }).Debug("server closed the connection normally") + case closeErr.Code == ws.CloseGoingAway: + // 1001 SERVER_SHUTDOWN: the server pod is restarting (routine deploy). + // The reconnect loop will land on a live pod, so don't alarm the user. c.cfg.Log.WithFields(log.Fields{ - "prefix": "hookdeckcli.ADDITIONAL_INFO", - }).Error("If you run into issues, please re-run with `--log-level debug` and share the output with the Hookdeck team on GitHub.") + "prefix": "websocket.Client.Close", + }).Debug("server is restarting, reconnecting: ", err) + case closeErr.Code == closeCodeSessionExpired: + // 4001 SESSION_EXPIRED: the session is gone from the server's store. + // Reconnecting recreates it via the X-Webhook-Ids / X-Session-Filters + // headers, so this is part of normal operation. + c.cfg.Log.WithFields(log.Fields{ + "prefix": "websocket.Client.Close", + }).Debug("session expired on server, reconnecting to recreate it: ", err) + case closeErr.Code == ws.CloseAbnormalClosure: + // 1006: the connection dropped without a close handshake — a network + // blip, laptop sleep, load balancer idle timeout, or a pod killed + // ungracefully. 1006 is never sent on the wire; gorilla synthesizes it + // for an unexpected EOF. The reconnect loop handles it, so this is + // routine rather than something to report. + c.cfg.Log.WithFields(log.Fields{ + "prefix": "websocket.Client.Close", + }).Debug("connection dropped, reconnecting: ", err) default: - c.cfg.Log.Error("other error: ", err) + c.cfg.Log.WithFields(log.Fields{ + "prefix": "websocket.Client.Close", + }).Error("close error: ", err) c.cfg.Log.WithFields(log.Fields{ "prefix": "hookdeckcli.ADDITIONAL_INFO", }).Error("If you run into issues, please re-run with `--log-level debug` and share the output with the Hookdeck team on GitHub.") @@ -433,7 +533,7 @@ func (c *Client) writePump() { // // NewClient returns a new Client. -func NewClient(url string, webSocketID string, CLIKey string, teamID string, cfg *Config) *Client { +func NewClient(url string, webSocketID string, CLIKey string, teamID string, connectionIDs []string, filtersJSON string, cfg *Config) *Client { if cfg == nil { cfg = &Config{} } @@ -469,11 +569,12 @@ func NewClient(url string, webSocketID string, CLIKey string, teamID string, cfg // Note that this client is not configured for websocket communications // and you must call c.changeConnection return &Client{ - URL: url, - WebSocketID: webSocketID, - // WebSocketAuthorizedFeature: websocketAuthorizedFeature, + URL: url, + WebSocketID: webSocketID, CLIKey: CLIKey, TeamID: teamID, + ConnectionIDs: connectionIDs, + FiltersJSON: filtersJSON, cfg: cfg, done: make(chan struct{}), send: make(chan *OutgoingMessage), @@ -491,6 +592,13 @@ const ( defaultPongWait = 10 * time.Second defaultWriteWait = 10 * time.Second + + // closeCodeSessionExpired (4001) is sent by the server when the CLI session + // no longer exists in its store, either on connect (older CLIs that don't + // send session-recreation headers) or mid-connection (session expired during + // a ping). Reconnecting with the X-Webhook-Ids / X-Session-Filters headers + // recreates the session, so this close code is expected, not an error. + closeCodeSessionExpired = 4001 ) // diff --git a/pkg/websocket/client_test.go b/pkg/websocket/client_test.go new file mode 100644 index 00000000..2381e6c7 --- /dev/null +++ b/pkg/websocket/client_test.go @@ -0,0 +1,258 @@ +package websocket + +import ( + "context" + "encoding/base64" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + ws "github.com/gorilla/websocket" + "github.com/sirupsen/logrus" + logtest "github.com/sirupsen/logrus/hooks/test" +) + +// upgradeTestServer starts an httptest server that upgrades websocket requests and captures +// the connect headers. The server-side connection is handed to onConn when provided. +func upgradeTestServer(t *testing.T, captured *http.Header, onConn func(conn *ws.Conn)) *httptest.Server { + t.Helper() + upgrader := ws.Upgrader{} + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *captured = r.Header.Clone() + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade failed: %v", err) + return + } + if onConn != nil { + onConn(conn) + } + })) +} + +func wsURL(s *httptest.Server) string { + return "ws" + strings.TrimPrefix(s.URL, "http") +} + +// startClient runs the client and waits for the websocket connection to be +// established. Tests must go through Run() rather than calling connect() +// directly: connect() starts the read/write pumps, and without Run's select +// loop draining notifyClose, readPump can block forever after Stop(). +func startClient(t *testing.T, client *Client) { + t.Helper() + go client.Run(context.Background()) + select { + case <-client.Connected(): + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the client to connect") + } +} + +func TestConnectSendsSessionRecreationHeaders(t *testing.T) { + filtersJSON := `{"body":{"name":"héllo wörld — テスト"}}` + var captured http.Header + server := upgradeTestServer(t, &captured, nil) + defer server.Close() + + client := NewClient( + wsURL(server), + "cses_test", + "cli-key", + "tm_test", + []string{"web_abc", "web_def"}, + filtersJSON, + &Config{}, + ) + startClient(t, client) + defer client.Stop() + + if got := captured.Get("Websocket-Id"); got != "cses_test" { + t.Errorf("Websocket-Id = %q, want %q", got, "cses_test") + } + if got := captured.Get("X-Webhook-Ids"); got != "web_abc,web_def" { + t.Errorf("X-Webhook-Ids = %q, want %q", got, "web_abc,web_def") + } + + encoded := captured.Get("X-Session-Filters") + if encoded == "" { + t.Fatal("X-Session-Filters header not sent") + } + // Base64 keeps the header value ASCII-safe: raw UTF-8 bytes would be decoded as latin-1 + // by the Node server and silently corrupt non-ASCII filter values. + for i := 0; i < len(encoded); i++ { + if encoded[i] > 127 { + t.Fatalf("X-Session-Filters contains non-ASCII byte at %d: %q", i, encoded) + } + } + decoded, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + t.Fatalf("X-Session-Filters is not valid base64: %v", err) + } + if string(decoded) != filtersJSON { + t.Errorf("decoded filters = %q, want %q", decoded, filtersJSON) + } +} + +func TestConnectOmitsSessionHeadersWhenUnset(t *testing.T) { + var captured http.Header + server := upgradeTestServer(t, &captured, nil) + defer server.Close() + + client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{}) + startClient(t, client) + defer client.Stop() + + if _, ok := captured["X-Webhook-Ids"]; ok { + t.Error("X-Webhook-Ids should not be sent when there are no connection IDs") + } + if _, ok := captured["X-Session-Filters"]; ok { + t.Error("X-Session-Filters should not be sent when there are no filters") + } +} + +func TestStopSendsCleanClose(t *testing.T) { + closeCode := make(chan int, 1) + var captured http.Header + server := upgradeTestServer(t, &captured, func(conn *ws.Conn) { + for { + _, _, err := conn.ReadMessage() + if err != nil { + if ce, ok := err.(*ws.CloseError); ok { + closeCode <- ce.Code + } else { + closeCode <- -1 + } + return + } + } + }) + defer server.Close() + + client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{}) + startClient(t, client) + + client.Stop() + + // The server must see a clean close (1000) — that's what lets it tombstone the session + // instead of holding it for the reconnect grace window. + select { + case code := <-closeCode: + if code != ws.CloseNormalClosure { + t.Errorf("server saw close code %d, want %d (normal closure)", code, ws.CloseNormalClosure) + } + case <-time.After(2 * time.Second): + t.Fatal("server did not receive a close frame") + } +} + +// The server intentionally closes with 1001 (pod restart during a deploy) and 4001 +// (session expired; recreated on reconnect via the session headers). Both are part of +// normal operation and must not produce error-level logs that alarm the user. +func TestServerCloseCodesReconnectQuietly(t *testing.T) { + codes := map[string]int{ + "server_shutdown_1001": ws.CloseGoingAway, + "session_expired_4001": closeCodeSessionExpired, + } + for name, code := range codes { + t.Run(name, func(t *testing.T) { + var captured http.Header + server := upgradeTestServer(t, &captured, func(conn *ws.Conn) { + _ = conn.WriteControl( + ws.CloseMessage, + ws.FormatCloseMessage(code, "test"), + time.Now().Add(time.Second), + ) + }) + defer server.Close() + + logger, hook := logtest.NewNullLogger() + client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{Log: logger}) + + go client.Run(context.Background()) + + select { + case <-client.NotifyExpired: + case <-time.After(5 * time.Second): + t.Fatal("client did not report connection loss after server close") + } + + if !client.HasConnected() { + t.Error("HasConnected() = false, want true after a successful connect that later dropped") + } + + for _, entry := range hook.AllEntries() { + if entry.Level <= logrus.ErrorLevel { + t.Errorf("close code %d logged at %s level: %s", code, entry.Level, entry.Message) + } + } + }) + } +} + +// A connection that dies without a close handshake (network blip, laptop sleep, load +// balancer timeout, ungracefully killed pod) surfaces as a synthesized 1006 CloseError. +// The reconnect loop handles it, so it must not tell the user to file a bug report. +func TestAbruptDisconnectReconnectsQuietly(t *testing.T) { + var captured http.Header + server := upgradeTestServer(t, &captured, func(conn *ws.Conn) { + // Drop the TCP connection with no close frame. + _ = conn.UnderlyingConn().Close() + }) + defer server.Close() + + logger, hook := logtest.NewNullLogger() + client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{Log: logger}) + + go client.Run(context.Background()) + + select { + case <-client.NotifyExpired: + case <-time.After(5 * time.Second): + t.Fatal("client did not report connection loss after the connection dropped") + } + + for _, entry := range hook.AllEntries() { + if entry.Level <= logrus.ErrorLevel { + t.Errorf("abrupt disconnect logged at %s level: %s", entry.Level, entry.Message) + } + } +} + +func TestStopIsIdempotentWithoutConnection(t *testing.T) { + client := NewClient("ws://127.0.0.1:1", "cses_test", "cli-key", "tm_test", nil, "", &Config{}) + + // Stop before any connection, twice: must not panic (doneOnce) and must close done. + client.Stop() + client.Stop() + + select { + case <-client.done: + default: + t.Fatal("done channel not closed after Stop") + } +} + +// Exercises the stateMu paths: Stop can run on the signal-handler goroutine while the connect +// goroutine writes conn/isConnected. Meaningful under `go test -race`. +func TestStopConcurrentWithConnect(t *testing.T) { + var captured http.Header + server := upgradeTestServer(t, &captured, nil) + defer server.Close() + + client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{}) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _ = client.connect(context.Background()) + }() + go func() { + defer wg.Done() + client.Stop() + }() + wg.Wait() +}