Added timeout for kupo client (#430)

Signed-off-by: Jenita <[email protected]>
Co-authored-by: Jenita <[email protected]>

Author: jkawan

input/chainsync/chainsync.go

@@ -22,6 +22,7 @@ import (
 	"log/slog"
 	"math"
 	"net/http"
+	"net/url"
 	"slices"
 	"strings"
 	"time"
@@ -43,6 +44,7 @@ const (
 	cursorCacheSize = 20
 
 	maxAutoReconnectDelay = 60 * time.Second
+	defaultKupoTimeout    = 30 * time.Second
 )
 
 type ChainSync struct {
@@ -507,41 +509,55 @@ func getKupoClient(c *ChainSync) (*kugo.Client, error) {
 		return c.kupoClient, nil
 	}
 
+	// Validate URL first
+	_, err := url.ParseRequestURI(c.kupoUrl)
+	if err != nil {
+		return nil, fmt.Errorf("invalid kupo URL: %w", err)
+	}
+
 	KugoCustomLogger := logging.NewKugoCustomLogger(logging.LevelInfo)
 
+	// Create client with timeout
 	k := kugo.New(
 		kugo.WithEndpoint(c.kupoUrl),
 		kugo.WithLogger(KugoCustomLogger),
+		kugo.WithTimeout(defaultKupoTimeout),
 	)
 
-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
+	httpClient := &http.Client{
+		Timeout: 2 * time.Second,
+	}
 
 	healthUrl := strings.TrimRight(c.kupoUrl, "/") + "/health"
+
+	// Create context with timeout
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthUrl, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create health check request: %w", err)
 	}
 
-	client := &http.Client{}
-	resp, err := client.Do(req)
+	resp, err := httpClient.Do(req)
 	if err != nil {
-		return nil, fmt.Errorf("failed to perform health check: %w", err)
-	}
-	if resp == nil {
-		return nil, errors.New("health check response empty, aborting")
+		// Handle different error types
+		switch {
+		case errors.Is(err, context.DeadlineExceeded):
+			return nil, errors.New("kupo health check timed out after 3 seconds")
+		case strings.Contains(err.Error(), "no such host"):
+			return nil, fmt.Errorf("failed to resolve kupo host: %w", err)
+		default:
+			return nil, fmt.Errorf("failed to perform health check: %w", err)
+		}
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf(
-			"health check failed with status code: %d",
-			resp.StatusCode,
-		)
+		return nil, fmt.Errorf("health check failed with status code: %d", resp.StatusCode)
 	}
 
 	c.kupoClient = k
-
 	return k, nil
 }
 
@@ -564,10 +580,18 @@ func resolveTransactionInputs(
 			// Extract transaction ID and index from the input
 			txId := input.Id().String()
 			txIndex := int(input.Index())
-			matches, err := k.Matches(context.Background(),
+
+			// Add timeout for matches query
+			ctx, cancel := context.WithTimeout(context.Background(), defaultKupoTimeout)
+			defer cancel()
+
+			matches, err := k.Matches(ctx,
 				kugo.TxOut(chainsync.NewTxID(txId, txIndex)),
 			)
 			if err != nil {
+				if errors.Is(err, context.DeadlineExceeded) {
+					return nil, fmt.Errorf("kupo matches query timed out after %v", defaultKupoTimeout)
+				}
 				return nil, fmt.Errorf(
 					"error fetching matches for input TxId: %s, Index: %d. Error: %w",
 					txId,

input/chainsync/chainsync_test.go

@@ -2,13 +2,19 @@ package chainsync
 
 import (
 	"encoding/hex"
+	//"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/SundaeSwap-finance/kugo"
 	"github.com/blinklabs-io/adder/event"
 	"github.com/blinklabs-io/gouroboros/protocol/chainsync"
 	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestHandleRollBackward(t *testing.T) {
@@ -68,3 +74,92 @@ func TestHandleRollBackward(t *testing.T) {
 	assert.Equal(t, uint64(67890), c.status.TipSlotNumber)
 	assert.Equal(t, "060708090a", c.status.TipBlockHash)
 }
+
+func TestGetKupoClient(t *testing.T) {
+	// Setup test server
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/health" {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	defer ts.Close()
+
+	t.Run("successful client creation", func(t *testing.T) {
+		c := &ChainSync{
+			kupoUrl: ts.URL,
+		}
+
+		client, err := getKupoClient(c)
+		require.NoError(t, err)
+		assert.NotNil(t, client)
+		assert.NotNil(t, c.kupoClient)
+	})
+
+	t.Run("returns cached client", func(t *testing.T) {
+		mockClient := &kugo.Client{}
+		c := &ChainSync{
+			kupoUrl:    ts.URL,
+			kupoClient: mockClient,
+		}
+
+		client, err := getKupoClient(c)
+		require.NoError(t, err)
+		assert.Same(t, mockClient, client)
+	})
+
+	t.Run("health check timeout", func(t *testing.T) {
+		slowTS := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(4 * time.Second) // Longer than the 3s context timeout
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer slowTS.Close()
+
+		c := &ChainSync{
+			kupoUrl: slowTS.URL,
+		}
+
+		_, err := getKupoClient(c)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "kupo health check timed out after 3 seconds")
+	})
+
+	t.Run("failed health check status", func(t *testing.T) {
+		failTS := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusInternalServerError)
+		}))
+		defer failTS.Close()
+
+		c := &ChainSync{
+			kupoUrl: failTS.URL,
+		}
+
+		_, err := getKupoClient(c)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "health check failed with status code: 500")
+	})
+
+	t.Run("malformed URL", func(t *testing.T) {
+		c := &ChainSync{
+			kupoUrl: "http://invalid url",
+		}
+
+		_, err := getKupoClient(c)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid kupo URL")
+	})
+
+	t.Run("unreachable host", func(t *testing.T) {
+		c := &ChainSync{
+			kupoUrl: "http://unreachable-host.invalid",
+		}
+
+		_, err := getKupoClient(c)
+		require.Error(t, err)
+		assert.True(t,
+			strings.Contains(err.Error(), "failed to resolve kupo host") ||
+				strings.Contains(err.Error(), "failed to perform health check"),
+			"unexpected error: %v", err)
+	})
+}