From 9ffa5624374a56f2122d7142d4fb8ab77a758f90 Mon Sep 17 00:00:00 2001 From: Troy Coombs Date: Thu, 16 Jul 2026 12:22:21 -0230 Subject: [PATCH 1/5] feat: send session data on connect and clean close on shutdown - Send X-Webhook-Ids and X-Session-Filters headers on every websocket connect/reconnect so the server can recreate the session in Redis if it expired between reconnects. - Send a clean websocket close (code 1000) on shutdown (Ctrl+C) so the server can tombstone the session immediately instead of holding it for the reconnect grace window. Stop() is now idempotent via sync.Once. Pairs with the server-side session tombstone in hookdeck/core#4477. Co-Authored-By: Claude Fable 5 --- pkg/listen/proxy/proxy.go | 28 +++++++++++++++++++ pkg/websocket/client.go | 57 ++++++++++++++++++++++++++++++++++----- 2 files changed, 78 insertions(+), 7 deletions(-) diff --git a/pkg/listen/proxy/proxy.go b/pkg/listen/proxy/proxy.go index 5437fae9..b9944a09 100644 --- a/pkg/listen/proxy/proxy.go +++ b/pkg/listen/proxy/proxy.go @@ -120,6 +120,15 @@ func (p *Proxy) Run(parentCtx context.Context) error { log.WithFields(log.Fields{ "prefix": "proxy.Proxy.Run", }).Debug("Ctrl+C received, cleaning up...") + + // 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 p.webSocketClient != nil { + p.webSocketClient.Stop() + } }) // Notify renderer we're connecting @@ -138,6 +147,20 @@ 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() { @@ -146,6 +169,8 @@ func (p *Proxy) Run(parentCtx context.Context) error { session.Id, p.cfg.Key, p.cfg.ProjectID, + connectionIDs, + filtersJSON, &websocket.Config{ Log: p.cfg.Log, NoWSS: p.cfg.NoWSS, @@ -182,6 +207,9 @@ func (p *Proxy) Run(parentCtx context.Context) error { // Block until ctrl+c, renderer quit, or websocket connection is interrupted select { case <-signalCtx.Done(): + if p.webSocketClient != nil { + p.webSocketClient.Stop() + } return nil case <-p.renderer.Done(): // Renderer wants to quit (user pressed q or similar) diff --git a/pkg/websocket/client.go b/pkg/websocket/client.go index 675a7dff..ca25c15f 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 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,6 +94,7 @@ type Client struct { conn *ws.Conn done chan struct{} + doneOnce sync.Once isConnected bool NotifyExpired chan struct{} @@ -162,9 +177,27 @@ 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() { + // 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 c.isConnected && c.conn != nil { + deadline := time.Now().Add(c.cfg.WriteWait) + _ = c.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 +255,15 @@ 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. + if len(c.ConnectionIDs) > 0 { + header.Set("X-Webhook-Ids", strings.Join(c.ConnectionIDs, ",")) + } + if c.FiltersJSON != "" { + header.Set("X-Session-Filters", c.FiltersJSON) + } + url := c.URL if c.cfg.NoWSS && strings.HasPrefix(url, "wss") { url = "ws" + strings.TrimPrefix(c.URL, "wss") @@ -433,7 +475,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 +511,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), From 10ec07c22519f8abe03fce33466ea25ef12c455a Mon Sep 17 00:00:00 2001 From: Troy Coombs Date: Fri, 17 Jul 2026 07:18:09 -0230 Subject: [PATCH 2/5] fix: guard websocket client state against cross-goroutine races; base64 filters header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Guard conn/isConnected with a mutex: Stop() runs on the signal-handler goroutine while the connect goroutine writes them, and the clean-close frame depends on reading a consistent snapshot. - Route all webSocketClient access through locked accessors: the reconnect loop reassigns it while the signal handler and event handlers read it. - Drop the Stop() call in the signalCtx.Done() branch — the signal callback always runs Stop() before cancelling the context. - Base64-encode the X-Session-Filters header: raw UTF-8 header bytes are decoded as latin-1 by the Node server and silently corrupt non-ASCII filter values on session recreation. - Add websocket client tests: recreation headers (incl. UTF-8 round-trip), clean close 1000 on Stop, Stop idempotency, and a race-detector exercise for concurrent Stop/connect. Co-Authored-By: Claude Fable 5 --- pkg/listen/proxy/proxy.go | 61 +++++++----- pkg/websocket/client.go | 40 ++++++-- pkg/websocket/client_test.go | 175 +++++++++++++++++++++++++++++++++++ 3 files changed, 247 insertions(+), 29 deletions(-) create mode 100644 pkg/websocket/client_test.go diff --git a/pkg/listen/proxy/proxy.go b/pkg/listen/proxy/proxy.go index b9944a09..bfe00755 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" @@ -66,10 +67,14 @@ type Config struct { // webhook events, forwards them to the local endpoint and sends the response // back to Hookdeck. type Proxy struct { - cfg *Config - connections []*hookdecksdk.Connection - webSocketClient *websocket.Client - connectionTimer *time.Timer + cfg *Config + connections []*hookdecksdk.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 @@ -98,6 +103,20 @@ func withSIGTERMCancel(ctx context.Context, onCancel func()) context.Context { // The connection is established in phases: // - Create a new CLI session // - Create a new websocket connection +// 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 +} + func (p *Proxy) Run(parentCtx context.Context) error { const maxConnectAttempts = 10 nAttempts := 0 @@ -126,8 +145,8 @@ func (p *Proxy) Run(parentCtx context.Context) error { // 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 p.webSocketClient != nil { - p.webSocketClient.Stop() + if wsClient := p.currentWebSocketClient(); wsClient != nil { + wsClient.Stop() } }) @@ -164,7 +183,7 @@ func (p *Proxy) Run(parentCtx context.Context) error { // 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, @@ -177,10 +196,11 @@ func (p *Proxy) Run(parentCtx context.Context) error { 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 @@ -201,24 +221,21 @@ 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(): - if p.webSocketClient != nil { - p.webSocketClient.Stop() - } + // 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: p.renderer.OnDisconnected() if !canConnect() { p.renderer.Cleanup() @@ -259,8 +276,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 @@ -405,7 +422,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{ @@ -432,7 +449,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{ @@ -472,8 +489,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 ca25c15f..87b0efed 100644 --- a/pkg/websocket/client.go +++ b/pkg/websocket/client.go @@ -82,7 +82,7 @@ type Client struct { ConnectionIDs []string // FiltersJSON is the JSON-encoded session filters (e.g., '{"body":{"action":"opened"}}'). - // Sent as the `X-Session-Filters` header on every connect/reconnect. + // Sent base64-encoded as the `X-Session-Filters` header on every connect/reconnect. // Empty string means no filters. FiltersJSON string @@ -96,6 +96,9 @@ type Client struct { 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 @@ -111,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) @@ -120,9 +123,21 @@ func (c *Client) Connected() <-chan struct{} { return d } +func (c *Client) connected() bool { + c.stateMu.Lock() + defer c.stateMu.Unlock() + return c.isConnected +} + +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") @@ -185,12 +200,19 @@ func (c *Client) ConnectionLost() { // instead of holding it open for the reconnect grace window. func (c *Client) Stop() { c.doneOnce.Do(func() { + // Snapshot the connection state under stateMu — Stop can run on the + // signal-handler goroutine while the connect goroutine writes these fields. + c.stateMu.Lock() + conn := c.conn + isConnected := c.isConnected + 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 c.isConnected && c.conn != nil { + if isConnected && conn != nil { deadline := time.Now().Add(c.cfg.WriteWait) - _ = c.conn.WriteControl( + _ = conn.WriteControl( ws.CloseMessage, ws.FormatCloseMessage(ws.CloseNormalClosure, "client_shutdown"), deadline, @@ -257,11 +279,13 @@ func (c *Client) connect(ctx context.Context) error { // 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", c.FiltersJSON) + header.Set("X-Session-Filters", base64.StdEncoding.EncodeToString([]byte(c.FiltersJSON))) } url := c.URL @@ -291,7 +315,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) @@ -309,7 +333,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{}) diff --git a/pkg/websocket/client_test.go b/pkg/websocket/client_test.go new file mode 100644 index 00000000..55e966ed --- /dev/null +++ b/pkg/websocket/client_test.go @@ -0,0 +1,175 @@ +package websocket + +import ( + "context" + "encoding/base64" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + ws "github.com/gorilla/websocket" +) + +// 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") +} + +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{}, + ) + if err := client.connect(context.Background()); err != nil { + t.Fatalf("connect failed: %v", err) + } + 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{}) + if err := client.connect(context.Background()); err != nil { + t.Fatalf("connect failed: %v", err) + } + 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{}) + if err := client.connect(context.Background()); err != nil { + t.Fatalf("connect failed: %v", err) + } + + 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") + } +} + +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() +} From 91509efaccd128b3271fc24fb100580552c8fe42 Mon Sep 17 00:00:00 2001 From: Troy Coombs Date: Tue, 4 Aug 2026 10:09:57 -0230 Subject: [PATCH 3/5] fix: reconnect quietly on server close codes 1001/4001; reset backoff after successful connections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Server deploys close CLI sockets with 1001 and session expiry closes with 4001 — both are routine and recovered by reconnecting (4001 recreation via the session headers), so stop logging them as errors. Also fixes the close dispatch bug where ws.IsCloseError(err) with no codes always returned false, leaving the error branches unreachable. Reset the attempt counter after a successful connection so backoff reflects consecutive failures instead of lifetime reconnects. Co-authored-by: Cursor --- pkg/listen/proxy/proxy.go | 26 +++++++++++++------- pkg/websocket/client.go | 39 +++++++++++++++++++++++++----- pkg/websocket/client_test.go | 46 ++++++++++++++++++++++++++++++++++++ 3 files changed, 96 insertions(+), 15 deletions(-) diff --git a/pkg/listen/proxy/proxy.go b/pkg/listen/proxy/proxy.go index bfe00755..03e01329 100644 --- a/pkg/listen/proxy/proxy.go +++ b/pkg/listen/proxy/proxy.go @@ -75,11 +75,11 @@ type Proxy struct { 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 + 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 @@ -99,10 +99,6 @@ func withSIGTERMCancel(ctx context.Context, onCancel func()) context.Context { return ctx } -// Run manages the connection to Hookdeck. -// The connection is established in phases: -// - Create a new CLI session -// - Create a new websocket connection // 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 { @@ -117,6 +113,10 @@ func (p *Proxy) setWebSocketClient(client *websocket.Client) { p.webSocketClient = client } +// Run manages the connection to Hookdeck. +// The connection is established in phases: +// - Create a new CLI session +// - Create a new websocket connection func (p *Proxy) Run(parentCtx context.Context) error { const maxConnectAttempts = 10 nAttempts := 0 @@ -237,6 +237,14 @@ func (p *Proxy) Run(parentCtx context.Context) error { return nil case <-wsClient.NotifyExpired: 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) diff --git a/pkg/websocket/client.go b/pkg/websocket/client.go index 87b0efed..5562f386 100644 --- a/pkg/websocket/client.go +++ b/pkg/websocket/client.go @@ -129,6 +129,13 @@ func (c *Client) connected() bool { 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 @@ -377,22 +384,35 @@ 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) 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.") @@ -560,6 +580,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 index 55e966ed..e16f98df 100644 --- a/pkg/websocket/client_test.go +++ b/pkg/websocket/client_test.go @@ -11,6 +11,8 @@ import ( "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 @@ -138,6 +140,50 @@ func TestStopSendsCleanClose(t *testing.T) { } } +// 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) + } + } + }) + } +} + func TestStopIsIdempotentWithoutConnection(t *testing.T) { client := NewClient("ws://127.0.0.1:1", "cses_test", "cli-key", "tm_test", nil, "", &Config{}) From 614179669d94270f6751f1318278e5825dc06174 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:35:00 +0000 Subject: [PATCH 4/5] =?UTF-8?q?fix:=20address=20Copilot=20review=20?= =?UTF-8?q?=E2=80=94=20clean=20close=20race=20in=20Stop,=20test=20goroutin?= =?UTF-8?q?e=20leaks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Stop(): key the clean-close frame on conn != nil alone. conn is only assigned after a successful upgrade, and also requiring isConnected skipped the close in the window between changeConnection() and setConnected(true). - Tests: go through Run() + Connected() instead of calling the unexported connect() directly, so readPump can't block forever on notifyClose after Stop() (goroutine leak). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MH9LQENoLJSawdD7X4yy2h --- pkg/websocket/client.go | 11 +++++++---- pkg/websocket/client_test.go | 26 +++++++++++++++++--------- 2 files changed, 24 insertions(+), 13 deletions(-) diff --git a/pkg/websocket/client.go b/pkg/websocket/client.go index 5562f386..4c3e3f7c 100644 --- a/pkg/websocket/client.go +++ b/pkg/websocket/client.go @@ -207,17 +207,20 @@ func (c *Client) ConnectionLost() { // instead of holding it open for the reconnect grace window. func (c *Client) Stop() { c.doneOnce.Do(func() { - // Snapshot the connection state under stateMu — Stop can run on the - // signal-handler goroutine while the connect goroutine writes these fields. + // 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 - isConnected := c.isConnected 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 isConnected && conn != nil { + if conn != nil { deadline := time.Now().Add(c.cfg.WriteWait) _ = conn.WriteControl( ws.CloseMessage, diff --git a/pkg/websocket/client_test.go b/pkg/websocket/client_test.go index e16f98df..5a6a8d32 100644 --- a/pkg/websocket/client_test.go +++ b/pkg/websocket/client_test.go @@ -37,6 +37,20 @@ 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 @@ -52,9 +66,7 @@ func TestConnectSendsSessionRecreationHeaders(t *testing.T) { filtersJSON, &Config{}, ) - if err := client.connect(context.Background()); err != nil { - t.Fatalf("connect failed: %v", err) - } + startClient(t, client) defer client.Stop() if got := captured.Get("Websocket-Id"); got != "cses_test" { @@ -90,9 +102,7 @@ func TestConnectOmitsSessionHeadersWhenUnset(t *testing.T) { defer server.Close() client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{}) - if err := client.connect(context.Background()); err != nil { - t.Fatalf("connect failed: %v", err) - } + startClient(t, client) defer client.Stop() if _, ok := captured["X-Webhook-Ids"]; ok { @@ -122,9 +132,7 @@ func TestStopSendsCleanClose(t *testing.T) { defer server.Close() client := NewClient(wsURL(server), "cses_test", "cli-key", "tm_test", nil, "", &Config{}) - if err := client.connect(context.Background()); err != nil { - t.Fatalf("connect failed: %v", err) - } + startClient(t, client) client.Stop() From cf9f0ff00bfe40657394070253300e08e4b69f18 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 14:09:51 +0000 Subject: [PATCH 5/5] fix(listen): quiet 1006 drops, no false reconnect on Ctrl+C, race-free connect flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three follow-ups from review of this branch: - Abnormal closure (1006) now logs at debug like 1001/4001. Fixing the latent ws.IsCloseError(err) bug revived the error-level branch, which meant an ordinary network blip, laptop sleep, LB idle timeout, or an ungracefully killed pod printed 'close error' plus an invitation to file a bug report. 1006 is never sent on the wire — gorilla synthesizes it for unexpected EOF — and the reconnect loop handles it. Covered by a regression test that drops the TCP connection with no close handshake. - Ctrl+C no longer announces 'Connection lost, reconnecting...' on the way out. Stopping the client closes its NotifyExpired channel, and because Stop runs before the context is cancelled, that case could win the select against signalCtx.Done() and start the reconnect spinner as the process exited (also leaving it running, since that path skips renderer.Cleanup). A shutdown flag set before Stop makes the intent unambiguous. - hasConnectedOnce is now an atomic.Bool. It was written by the per-attempt connection monitor goroutine and read by canConnect on the Run goroutine — a data race that -race never caught because this package has no tests. CompareAndSwap also makes the 'spawn the health monitor exactly once' guarantee real. Co-Authored-By: Claude --- pkg/listen/proxy/proxy.go | 33 +++++++++++++++++++++++++-------- pkg/websocket/client.go | 9 +++++++++ pkg/websocket/client_test.go | 29 +++++++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 8 deletions(-) diff --git a/pkg/listen/proxy/proxy.go b/pkg/listen/proxy/proxy.go index f0671591..fd16a656 100644 --- a/pkg/listen/proxy/proxy.go +++ b/pkg/listen/proxy/proxy.go @@ -129,20 +129,30 @@ 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 @@ -207,12 +217,11 @@ func (p *Proxy) Run(parentCtx context.Context) error { 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 @@ -245,6 +254,14 @@ func (p *Proxy) Run(parentCtx context.Context) error { p.renderer.Cleanup() return nil 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 diff --git a/pkg/websocket/client.go b/pkg/websocket/client.go index 4c3e3f7c..505e1611 100644 --- a/pkg/websocket/client.go +++ b/pkg/websocket/client.go @@ -412,6 +412,15 @@ func (c *Client) readPump() { 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.WithFields(log.Fields{ "prefix": "websocket.Client.Close", diff --git a/pkg/websocket/client_test.go b/pkg/websocket/client_test.go index 5a6a8d32..2381e6c7 100644 --- a/pkg/websocket/client_test.go +++ b/pkg/websocket/client_test.go @@ -192,6 +192,35 @@ func TestServerCloseCodesReconnectQuietly(t *testing.T) { } } +// 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{})