Merge pull request #22 from codeGROOVE-dev/star-org

Improve client reliability

Author: tstromberg

pkg/client/client.go

@@ -32,6 +32,10 @@ const (
 	// UI constants for logging.
 	separatorLine = "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
 	msgTypeField  = "type"
+
+	// Server ping timing constants.
+	expectedPingInterval = 54 * time.Second
+	pingDelayThreshold   = 10 * time.Second
 )
 
 // Event represents a webhook event received from the server.
@@ -63,14 +67,16 @@ type Config struct {
 
 // Client represents a WebSocket client with automatic reconnection.
 type Client struct {
-	logger     *slog.Logger
-	ws         *websocket.Conn
-	stopCh     chan struct{}
-	stoppedCh  chan struct{}
-	config     Config
-	eventCount int
-	retries    int
-	mu         sync.RWMutex
+	logger          *slog.Logger
+	ws              *websocket.Conn
+	stopCh          chan struct{}
+	stoppedCh       chan struct{}
+	config          Config
+	lastPingTime    time.Time // Time of last ping received
+	noActivitySince time.Time // Track when we last received anything
+	mu              sync.RWMutex
+	eventCount      int
+	retries         int
 }
 
 // New creates a new robust WebSocket client.
@@ -454,26 +460,62 @@ func (c *Client) readEvents(ctx context.Context, ws *websocket.Conn) error {
 
 		// Handle ping messages
 		if responseType == "ping" {
+			now := time.Now()
+
+			// Check if pings are arriving on schedule
+			c.mu.Lock()
+			lastPing := c.lastPingTime
+			c.lastPingTime = now
+			c.noActivitySince = now
+			c.mu.Unlock()
+
+			if !lastPing.IsZero() {
+				timeSinceLastPing := now.Sub(lastPing)
+				if timeSinceLastPing > expectedPingInterval+pingDelayThreshold {
+					c.logger.Warn("[PING-PONG] Delayed ping from server",
+						"expected_interval", expectedPingInterval,
+						"actual_interval", timeSinceLastPing,
+						"delay", timeSinceLastPing-expectedPingInterval)
+				}
+			}
+
 			c.logger.Debug("[PING-PONG] Received PING from server")
 			pong := map[string]any{msgTypeField: "pong"}
 			// Echo back any sequence number if present
 			if seq, ok := response["seq"]; ok {
 				pong["seq"] = seq
 			}
+			// Set write deadline to ensure timely pong response
+			if err := ws.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+				c.logger.Error("[PING-PONG] Failed to set write deadline", "error", err)
+				return fmt.Errorf("error setting write deadline: %w", err)
+			}
 			if err := websocket.JSON.Send(ws, pong); err != nil {
 				c.logger.Error("[PING-PONG] Failed to send PONG response", "error", err)
 				return fmt.Errorf("error sending pong response: %w", err)
 			}
+			// Clear write deadline
+			if err := ws.SetWriteDeadline(time.Time{}); err != nil {
+				c.logger.Warn("[PING-PONG] Failed to clear write deadline", "error", err)
+			}
 			c.logger.Debug("[PING-PONG] Sent PONG response to server", "seq", pong["seq"])
 			continue
 		}
 
 		// Handle pong acknowledgments
 		if responseType == "pong" {
+			c.mu.Lock()
+			c.noActivitySince = time.Now()
+			c.mu.Unlock()
 			c.logger.Debug("[PING-PONG] Received PONG acknowledgment from server")
 			continue
 		}
 
+		// Update activity timestamp for any event
+		c.mu.Lock()
+		c.noActivitySince = time.Now()
+		c.mu.Unlock()
+
 		// Process the event inline
 		event := Event{
 			Type: responseType,

pkg/hub/client.go

@@ -13,15 +13,19 @@ import (
 
 // Client represents a connected WebSocket client with their subscription preferences.
 type Client struct {
-	conn         *websocket.Conn
-	send         chan Event
-	hub          *Hub
-	done         chan struct{}
-	userOrgs     map[string]bool
-	ID           string
-	subscription Subscription
-	closeOnce    sync.Once
-	pingSeq      int64 // Track ping sequence numbers
+	conn            *websocket.Conn
+	send            chan Event
+	hub             *Hub
+	done            chan struct{}
+	userOrgs        map[string]bool
+	ID              string
+	subscription    Subscription
+	lastPongTime    time.Time // Time of last pong
+	mu              sync.RWMutex
+	closeOnce       sync.Once
+	pingSeq         int64 // Track ping sequence numbers
+	lastPongSeq     int64 // Last pong sequence received
+	missedPongCount int   // Count of consecutive missed pongs
 }
 
 // NewClient creates a new client.
@@ -101,16 +105,35 @@ func (c *Client) Run(ctx context.Context, pingInterval, writeTimeout time.Durati
 			log.Printf("✓ Event sent to client %s", c.ID)
 
 		case <-ticker.C:
+			// Check if we're missing pongs before sending next ping
+			c.mu.RLock()
+			currentSeq := c.pingSeq
+			lastPongSeq := c.lastPongSeq
+			c.mu.RUnlock()
+
+			if currentSeq > 0 && lastPongSeq < currentSeq {
+				c.mu.Lock()
+				c.missedPongCount++
+				missedCount := c.missedPongCount
+				c.mu.Unlock()
+				log.Printf("WARNING: client %s has not responded to ping #%d (missed %d consecutive pongs)",
+					c.ID, currentSeq, missedCount)
+			}
+
 			// Send ping to keep connection alive
+			c.mu.Lock()
 			c.pingSeq++
+			currentSeq = c.pingSeq
+			c.mu.Unlock()
+
 			if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
 				log.Printf("error setting ping deadline for client %s: %v", c.ID, err)
 				return
 			}
 			ping := map[string]any{
 				"type":      "ping",
 				"timestamp": time.Now().Format(time.RFC3339),
-				"seq":       c.pingSeq,
+				"seq":       currentSeq,
 			}
 			if err := websocket.JSON.Send(c.conn, ping); err != nil {
 				log.Printf("error sending ping to client %s: %v", c.ID, err)
@@ -125,6 +148,16 @@ func (c *Client) Run(ctx context.Context, pingInterval, writeTimeout time.Durati
 	}
 }
 
+// RecordPong records receipt of a pong from the client.
+func (c *Client) RecordPong(seq int64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.lastPongSeq = seq
+	c.lastPongTime = time.Now()
+	c.missedPongCount = 0
+}
+
 // Close gracefully closes the client.
 func (c *Client) Close() {
 	c.closeOnce.Do(func() {

pkg/hub/websocket.go

@@ -22,7 +22,7 @@ import (
 // Constants for WebSocket timeouts and limits.
 const (
 	pingInterval        = 54 * time.Second
-	readTimeout         = 60 * time.Second // Must be > pingInterval to avoid false timeouts
+	readTimeout         = 90 * time.Second // Must be > pingInterval + response time to avoid false timeouts
 	writeTimeout        = 10 * time.Second
 	minTokenLength      = 40   // Minimum GitHub token length
 	maxTokenLength      = 255  // Maximum GitHub token length
@@ -717,6 +717,9 @@ func (h *WebSocketHandler) Handle(ws *websocket.Conn) {
 				switch msgType {
 				case "pong":
 					// Pong received - connection is alive
+					if seq, ok := msgMap["seq"].(float64); ok {
+						client.RecordPong(int64(seq))
+					}
 					continue
 				case "ping":
 					// Client sent us a ping, send pong back