feat: implement Direct RPC mode for Smart Router

protocol/chainlib/chainlib.go

@@ -15,6 +15,7 @@ import (
 	"github.com/lavanet/lava/v5/utils"
 	pairingtypes "github.com/lavanet/lava/v5/x/pairing/types"
 	spectypes "github.com/lavanet/lava/v5/x/spec/types"
+	"google.golang.org/grpc"
 )
 
 const (
@@ -42,7 +43,7 @@ func NewChainListener(
 	healthReporter HealthReporter,
 	rpcConsumerLogs *metrics.RPCConsumerLogs,
 	chainParser ChainParser,
-	consumerWsSubscriptionManager *ConsumerWSSubscriptionManager,
+	consumerWsSubscriptionManager WSSubscriptionManager,
 ) (ChainListener, error) {
 	if listenEndpoint.NetworkAddress == INTERNAL_ADDRESS {
 		utils.LavaFormatDebug("skipping chain listener for internal address")
@@ -118,6 +119,16 @@ type HealthReporter interface {
 	IsHealthy() bool
 }
 
+// GRPCReflectionProvider is an optional interface that can be implemented by RelaySender
+// to provide gRPC reflection support. When implemented, the gRPC listener will register
+// a reflection proxy service that enables tools like grpcurl to discover services.
+type GRPCReflectionProvider interface {
+	// GetGRPCReflectionConnection returns a gRPC connection for reflection requests.
+	// The cleanup function should be called when the connection is no longer needed.
+	// Returns nil if reflection is not supported.
+	GetGRPCReflectionConnection(ctx context.Context) (conn *grpc.ClientConn, cleanup func(), err error)
+}
+
 type RelaySender interface {
 	SendRelay(
 		ctx context.Context,

protocol/chainlib/chainproxy/rpcInterfaceMessages/restMessage.go

@@ -1,6 +1,7 @@
 package rpcInterfaceMessages
 
 import (
+	"fmt"
 	"net/url"
 	"strings"
 
@@ -49,6 +50,13 @@ func (rm *RestMessage) GetRawRequestHash() ([]byte, error) {
 }
 
 func (jm RestMessage) CheckResponseError(data []byte, httpStatusCode int) (hasError bool, errorMessage string) {
+	// Treat 5xx and 429 as node errors (triggers retries)
+	if httpStatusCode >= 500 || httpStatusCode == 429 {
+		// Server error or rate limit - treat as node error for retry logic
+		errorMsg := extractErrorMessage(data, httpStatusCode)
+		return true, errorMsg
+	}
+
 	// Check Cosmos SDK transaction errors (HTTP 2xx with error code in JSON)
 	if httpStatusCode >= 200 && httpStatusCode < 300 {
 		if hasError, errMsg := checkCosmosTxError(data); hasError {
@@ -58,8 +66,9 @@ func (jm RestMessage) CheckResponseError(data []byte, httpStatusCode int) (hasEr
 		return false, ""
 	}
 
-	// Check generic REST errors (non-2xx HTTP status)
-	return checkGenericRESTError(data)
+	// 4xx (except 429) are client errors - NOT node errors
+	// Return false so state machine doesn't retry (client error won't succeed on retry)
+	return false, ""
 }
 
 // checkCosmosTxError detects errors in Cosmos SDK transaction responses
@@ -78,7 +87,41 @@ func checkCosmosTxError(data []byte) (bool, string) {
 	return false, ""
 }
 
-// checkGenericRESTError detects errors in generic REST API responses
+// extractErrorMessage attempts to extract error message from response body
+// Tries common fields in order: "message", "error", raw body (truncated), or fallback to status
+func extractErrorMessage(data []byte, httpStatusCode int) string {
+	// Try to parse as JSON and extract error message
+	var result map[string]interface{}
+	if err := json.Unmarshal(data, &result); err == nil {
+		// Try common error field names
+		if msg, ok := result["message"].(string); ok && msg != "" {
+			return msg
+		}
+		if msg, ok := result["error"].(string); ok && msg != "" {
+			return msg
+		}
+		// Try nested error.message
+		if errorObj, ok := result["error"].(map[string]interface{}); ok {
+			if msg, ok := errorObj["message"].(string); ok && msg != "" {
+				return msg
+			}
+		}
+	}
+
+	// Fallback: use raw body (truncated to 1KB)
+	bodyStr := string(data)
+	if len(bodyStr) > 1024 {
+		bodyStr = bodyStr[:1024] + "..."
+	}
+	if bodyStr != "" {
+		return bodyStr
+	}
+
+	// Final fallback: use HTTP status code
+	return fmt.Sprintf("HTTP %d", httpStatusCode)
+}
+
+// checkGenericRESTError detects errors in generic REST API responses (deprecated - kept for backward compatibility)
 // Expects format: {"message": "error text", "code": <error_code>}
 func checkGenericRESTError(data []byte) (bool, string) {
 	result := make(map[string]interface{})

protocol/chainlib/chainproxy/rpcInterfaceMessages/restMessage_test.go

@@ -541,3 +541,93 @@
 		})
 	}
 }
+
+// TestCheckResponseError_ServerErrors tests 5xx and 429 handling (Phase 4 corrections)
+func TestCheckResponseError_ServerErrors(t *testing.T) {
+	testCases := []struct {
+		name          string
+		httpStatus    int
+		response      string
+		expectedError bool  // Should be node error?
+		errorCheck    func(t *testing.T, errorMessage string)
+	}{
+		{
+			name:          "503 Service Unavailable with JSON error",
+			httpStatus:    503,
+			response:      `{"error":"service temporarily unavailable"}`,
+			expectedError: true,  // Node error - triggers retry
+			errorCheck: func(t *testing.T, errorMessage string) {
+				require.Contains(t, errorMessage, "service temporarily unavailable")
+			},
+		},
+		{
+			name:          "503 with plain text body",
+			httpStatus:    503,
+			response:      `Service Unavailable`,
+			expectedError: true,  // Node error even without JSON
+			errorCheck: func(t *testing.T, errorMessage string) {
+				require.Contains(t, errorMessage, "Service Unavailable")
+			},
+		},
+		{
+			name:          "500 Internal Server Error",
+			httpStatus:    500,
+			response:      `{"message":"internal server error"}`,
+			expectedError: true,  // Node error
+			errorCheck: func(t *testing.T, errorMessage string) {
+				require.Contains(t, errorMessage, "internal server error")
+			},
+		},
+		{
+			name:          "502 Bad Gateway",
+			httpStatus:    502,
+			response:      `Bad Gateway`,
+			expectedError: true,  // Node error
+		},
+		{
+			name:          "429 Rate Limit Exceeded",
+			httpStatus:    429,
+			response:      `{"error":"rate limit exceeded"}`,
+			expectedError: true,  // Node error (triggers backoff/retry)
+			errorCheck: func(t *testing.T, errorMessage string) {
+				require.Contains(t, errorMessage, "rate limit exceeded")
+			},
+		},
+		{
+			name:          "429 with empty body",
+			httpStatus:    429,
+			response:      ``,
+			expectedError: true,  // Still node error
+			errorCheck: func(t *testing.T, errorMessage string) {
+				require.Equal(t, "HTTP 429", errorMessage)  // Fallback to status
+			},
+		},
+		{
+			name:          "404 Not Found (client error)",
+			httpStatus:    404,
+			response:      `{"code":5,"message":"block not found"}`,
+			expectedError: false,  // NOT a node error - client error
+		},
+		{
+			name:          "400 Bad Request (client error)",
+			httpStatus:    400,
+			response:      `{"error":"invalid request"}`,
+			expectedError: false,  // NOT a node error
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			restMsg := RestMessage{}
+
+			hasError, errorMessage := restMsg.CheckResponseError([]byte(tc.response), tc.httpStatus)
+
+			require.Equal(t, tc.expectedError, hasError,
+				"Expected node error=%v for HTTP %d", tc.expectedError, tc.httpStatus)
+
+			if tc.errorCheck != nil {
+				tc.errorCheck(t, errorMessage)
+			}
+		})
+	}
+}

protocol/chainlib/chainproxy/rpcclient/subscription_test.go

@@ -42,7 +42,7 @@ func TestClientSubscription_UnsubscribeBeforeRun(t *testing.T) {
 	select {
 	case <-done:
 		// Success - Unsubscribe completed without blocking
-		t.Log("✓ Unsubscribe completed without deadlock")
+		t.Log("[PASS] Unsubscribe completed without deadlock")
 	case <-time.After(1 * time.Second):
 		t.Fatal("Unsubscribe deadlocked when called before run() started")
 	}
@@ -87,7 +87,7 @@ func TestClientSubscription_UnsubscribeRace(t *testing.T) {
 
 	select {
 	case <-done:
-		t.Log("✓ Concurrent Unsubscribe calls completed safely")
+		t.Log("[PASS] Concurrent Unsubscribe calls completed safely")
 	case <-time.After(2 * time.Second):
 		t.Fatal("Concurrent Unsubscribe calls deadlocked")
 	}
@@ -127,7 +127,7 @@ func TestClientSubscription_UnsubscribeWithRunning(t *testing.T) {
 
 	select {
 	case <-done:
-		t.Log("✓ Subscription run() completed after Unsubscribe")
+		t.Log("[PASS] Subscription run() completed after Unsubscribe")
 	case <-time.After(2 * time.Second):
 		t.Fatal("Subscription run() didn't exit after Unsubscribe")
 	}
@@ -168,7 +168,7 @@ func TestClientSubscription_BufferedQuitChannelPreventsDeadlock(t *testing.T) {
 	// This is the key test: with buffered channel, this should succeed
 	select {
 	case <-unsubscribeComplete:
-		t.Log("✓ Buffered quit channel prevented deadlock")
+		t.Log("[PASS] Buffered quit channel prevented deadlock")
 
 		// Now verify the quit signal is still available if run() starts later
 		select {

protocol/chainlib/consumer_websocket_manager.go

@@ -38,7 +38,7 @@ type ConsumerWebsocketManager struct {
 	apiInterface                  string
 	connectionType                string
 	relaySender                   RelaySender
-	consumerWsSubscriptionManager *ConsumerWSSubscriptionManager
+	consumerWsSubscriptionManager WSSubscriptionManager
 	WebsocketConnectionUID        string
 	headerRateLimit               uint64
 }
@@ -52,7 +52,7 @@ type ConsumerWebsocketManagerOptions struct {
 	ApiInterface                  string
 	ConnectionType                string
 	RelaySender                   RelaySender
-	ConsumerWsSubscriptionManager *ConsumerWSSubscriptionManager
+	ConsumerWsSubscriptionManager WSSubscriptionManager
 	WebsocketConnectionUID        string
 	headerRateLimit               uint64
 }

protocol/chainlib/grpc.go

@@ -311,7 +311,16 @@ func (apil *GrpcChainListener) Serve(ctx context.Context, cmdFlags common.Consum
 		return relayReply.Data, convertRelayMetaDataToMDMetaData(metadataToReply), nil
 	}
 
-	_, httpServer, err := grpcproxy.NewGRPCProxy(sendRelayCallback, apil.endpoint.HealthCheckPath, cmdFlags, apil.healthReporter)
+	// Check if the relay sender supports gRPC reflection (optional interface)
+	var reflectionCallback grpcproxy.ReflectionProxyCallback
+	if reflectionProvider, ok := apil.relaySender.(GRPCReflectionProvider); ok {
+		reflectionCallback = reflectionProvider.GetGRPCReflectionConnection
+		utils.LavaFormatInfo("gRPC reflection support enabled",
+			utils.LogAttr("address", apil.endpoint.NetworkAddress),
+		)
+	}
+
+	_, httpServer, err := grpcproxy.NewGRPCProxyWithReflection(sendRelayCallback, apil.endpoint.HealthCheckPath, cmdFlags, apil.healthReporter, reflectionCallback)
 	if err != nil {
 		utils.LavaFormatFatal("provider failure RegisterServer", err, utils.Attribute{Key: "listenAddr", Value: apil.endpoint.NetworkAddress})
 	}