diff --git a/internal/tempproxy/client.go b/internal/tempproxy/client.go index 08fa63a..7bd87d8 100644 --- a/internal/tempproxy/client.go +++ b/internal/tempproxy/client.go @@ -1,8 +1,11 @@ package tempproxy import ( + "context" "encoding/json" "fmt" + "math" + "math/rand" "net" "os" "time" @@ -12,8 +15,46 @@ const ( defaultSocketPath = "/run/pulse-sensor-proxy/pulse-sensor-proxy.sock" containerSocketPath = "/mnt/pulse-proxy/pulse-sensor-proxy.sock" defaultTimeout = 30 * time.Second // Increased to accommodate SSH operations + + // Exponential backoff constants + initialBackoff = 100 * time.Millisecond + maxBackoff = 10 * time.Second + backoffFactor = 2.0 + jitterFraction = 0.1 + maxRetries = 3 ) +// ErrorType classifies proxy errors for better error handling +type ErrorType int + +const ( + ErrorTypeUnknown ErrorType = iota + ErrorTypeTransport // Socket connection/communication failures + ErrorTypeAuth // Authorization failures + ErrorTypeSSH // SSH connectivity issues + ErrorTypeSensor // Sensor command failures + ErrorTypeTimeout // Operation timeout +) + +// ProxyError wraps errors with classification +type ProxyError struct { + Type ErrorType + Message string + Retryable bool + Wrapped error +} + +func (e *ProxyError) Error() string { + if e.Wrapped != nil { + return fmt.Sprintf("%s: %v", e.Message, e.Wrapped) + } + return e.Message +} + +func (e *ProxyError) Unwrap() error { + return e.Wrapped +} + // Client communicates with pulse-sensor-proxy via unix socket type Client struct { socketPath string @@ -58,17 +99,213 @@ type RPCResponse struct { Error string `json:"error,omitempty"` } -// call sends an RPC request and returns the response -func (c *Client) call(method string, params map[string]interface{}) (*RPCResponse, error) { - // Connect to unix socket - conn, err := net.DialTimeout("unix", c.socketPath, c.timeout) +// calculateBackoff calculates exponential backoff with jitter +func calculateBackoff(attempt int) time.Duration { + if attempt <= 0 { + return initialBackoff + } + + // Calculate base backoff: initialBackoff * (backoffFactor ^ attempt) + backoff := float64(initialBackoff) * math.Pow(backoffFactor, float64(attempt)) + + // Cap at maxBackoff + if backoff > float64(maxBackoff) { + backoff = float64(maxBackoff) + } + + // Add jitter: ±10% randomization + jitter := backoff * jitterFraction * (rand.Float64()*2 - 1) + backoff += jitter + + return time.Duration(backoff) +} + +// classifyError categorizes errors for retry logic +func classifyError(err error, respError string) *ProxyError { + if err == nil { + return nil + } + + // Check for timeout + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + return &ProxyError{ + Type: ErrorTypeTimeout, + Message: "operation timed out", + Retryable: true, + Wrapped: err, + } + } + + // Check for connection errors (socket unavailable) + if _, ok := err.(*net.OpError); ok { + return &ProxyError{ + Type: ErrorTypeTransport, + Message: "failed to connect to proxy socket", + Retryable: true, + Wrapped: err, + } + } + + // Check response error messages + if respError != "" { + if respError == "unauthorized" || respError == "method requires host-level privileges" { + return &ProxyError{ + Type: ErrorTypeAuth, + Message: respError, + Retryable: false, + Wrapped: fmt.Errorf("%s", respError), + } + } + + // SSH-related errors + if contains(respError, "ssh", "connection", "timeout") { + return &ProxyError{ + Type: ErrorTypeSSH, + Message: "SSH connectivity issue", + Retryable: true, + Wrapped: fmt.Errorf("%s", respError), + } + } + + // Sensor errors + if contains(respError, "sensor", "temperature") { + return &ProxyError{ + Type: ErrorTypeSensor, + Message: "sensor command failed", + Retryable: false, + Wrapped: fmt.Errorf("%s", respError), + } + } + } + + // Unknown error + return &ProxyError{ + Type: ErrorTypeUnknown, + Message: "unknown proxy error", + Retryable: false, + Wrapped: err, + } +} + +// contains checks if any of the substrings are in the main string (case-insensitive) +func contains(s string, substrs ...string) bool { + s = fmt.Sprintf("%v", s) + for _, substr := range substrs { + if len(s) >= len(substr) { + for i := 0; i <= len(s)-len(substr); i++ { + match := true + for j := 0; j < len(substr); j++ { + c1 := s[i+j] + c2 := substr[j] + if c1 >= 'A' && c1 <= 'Z' { + c1 += 32 + } + if c2 >= 'A' && c2 <= 'Z' { + c2 += 32 + } + if c1 != c2 { + match = false + break + } + } + if match { + return true + } + } + } + } + return false +} + +// callWithContext sends an RPC request with context and retry support +func (c *Client) callWithContext(ctx context.Context, method string, params map[string]interface{}) (*RPCResponse, error) { + var lastErr error + + for attempt := 0; attempt <= maxRetries; attempt++ { + // Check if context is already cancelled + select { + case <-ctx.Done(): + return nil, &ProxyError{ + Type: ErrorTypeTimeout, + Message: "context cancelled before retry", + Retryable: false, + Wrapped: ctx.Err(), + } + default: + } + + // Try the call + resp, err := c.callOnce(ctx, method, params) + + // Success + if err == nil && resp != nil && resp.Success { + return resp, nil + } + + // Classify error + respError := "" + if resp != nil { + respError = resp.Error + } + proxyErr := classifyError(err, respError) + + // Don't retry non-retryable errors + if proxyErr != nil && !proxyErr.Retryable { + return resp, proxyErr + } + + lastErr = proxyErr + if lastErr == nil { + lastErr = err + } + + // Don't sleep after last attempt + if attempt < maxRetries { + backoff := calculateBackoff(attempt) + + select { + case <-time.After(backoff): + // Continue to next attempt + case <-ctx.Done(): + return nil, &ProxyError{ + Type: ErrorTypeTimeout, + Message: "context cancelled during backoff", + Retryable: false, + Wrapped: ctx.Err(), + } + } + } + } + + // All retries exhausted + return nil, &ProxyError{ + Type: ErrorTypeTransport, + Message: fmt.Sprintf("max retries (%d) exhausted", maxRetries), + Retryable: false, + Wrapped: lastErr, + } +} + +// callOnce sends a single RPC request without retries +func (c *Client) callOnce(ctx context.Context, method string, params map[string]interface{}) (*RPCResponse, error) { + // Create a dialer with context + dialer := net.Dialer{ + Timeout: c.timeout, + } + + // Connect to unix socket with context + conn, err := dialer.DialContext(ctx, "unix", c.socketPath) if err != nil { return nil, fmt.Errorf("failed to connect to proxy: %w", err) } defer conn.Close() - // Set deadline - conn.SetDeadline(time.Now().Add(c.timeout)) + // Set deadline from context or use default timeout + deadline, ok := ctx.Deadline() + if !ok { + deadline = time.Now().Add(c.timeout) + } + conn.SetDeadline(deadline) // Send request req := RPCRequest{ @@ -81,7 +318,7 @@ func (c *Client) call(method string, params map[string]interface{}) (*RPCRespons return nil, fmt.Errorf("failed to encode request: %w", err) } - // Read response (server uses newline-delimited framing, no CloseWrite needed) + // Read response (server uses newline-delimited framing) var resp RPCResponse decoder := json.NewDecoder(conn) if err := decoder.Decode(&resp); err != nil { @@ -91,6 +328,13 @@ func (c *Client) call(method string, params map[string]interface{}) (*RPCRespons return &resp, nil } +// call sends an RPC request and returns the response (legacy method, uses default context) +func (c *Client) call(method string, params map[string]interface{}) (*RPCResponse, error) { + ctx, cancel := context.WithTimeout(context.Background(), c.timeout) + defer cancel() + return c.callWithContext(ctx, method, params) +} + // GetStatus returns proxy status func (c *Client) GetStatus() (map[string]interface{}, error) { resp, err := c.call("get_status", nil)