feat: PRT-improve-timeout-handling

ecosystem/cache/handlers.go

@@ -126,16 +126,17 @@ func (s *RelayerCacheServer) GetRelay(ctx context.Context, relayCacheGet *pairin
 		}()
 		// wait for all reads to complete before moving forward
 		waitGroup.Wait()
-
-		// validate that the response seen block is larger or equal to our expectations.
-		if cacheReply.SeenBlock < slices.Min([]int64{relayCacheGet.SeenBlock, relayCacheGet.RequestedBlock}) { // TODO unitest this.
-			// Error, our reply seen block is not larger than our expectations, meaning we got an old response
-			// this can happen only in the case relayCacheGet.SeenBlock < relayCacheGet.RequestedBlock
-			// by setting the err variable we will get a cache miss, and the relay will continue to the node.
-			err = utils.LavaFormatDebug("reply seen block is smaller than our expectations",
-				utils.LogAttr("cacheReply.SeenBlock", cacheReply.SeenBlock),
-				utils.LogAttr("seenBlock", relayCacheGet.SeenBlock),
-			)
+		if err == nil { // in case we got a hit validate seen block of the reply.
+			// validate that the response seen block is larger or equal to our expectations.
+			if cacheReply.SeenBlock < slices.Min([]int64{relayCacheGet.SeenBlock, relayCacheGet.RequestedBlock}) { // TODO unitest this.
+				// Error, our reply seen block is not larger than our expectations, meaning we got an old response
+				// this can happen only in the case relayCacheGet.SeenBlock < relayCacheGet.RequestedBlock
+				// by setting the err variable we will get a cache miss, and the relay will continue to the node.
+				err = utils.LavaFormatDebug("reply seen block is smaller than our expectations",
+					utils.LogAttr("cacheReply.SeenBlock", cacheReply.SeenBlock),
+					utils.LogAttr("seenBlock", relayCacheGet.SeenBlock),
+				)
+			}
 		}
 		// set seen block.
 		if relayCacheGet.SeenBlock > cacheReply.SeenBlock {

protocol/chainlib/chain_message_queries.go

@@ -3,25 +3,25 @@ package chainlib
 import "github.com/lavanet/lava/protocol/common"
 
 func ShouldSendToAllProviders(chainMessage ChainMessage) bool {
-	return chainMessage.GetApi().Category.Stateful == common.CONSISTENCY_SELECT_ALLPROVIDERS
+	return chainMessage.GetApi().Category.Stateful == common.CONSISTENCY_SELECT_ALL_PROVIDERS
 }
 
-func GetAddon(chainMessage ChainMessage) string {
+func GetAddon(chainMessage ChainMessageForSend) string {
 	return chainMessage.GetApiCollection().CollectionData.AddOn
 }
 
-func IsSubscription(chainMessage ChainMessage) bool {
+func IsSubscription(chainMessage ChainMessageForSend) bool {
 	return chainMessage.GetApi().Category.Subscription
 }
 
-func IsHangingApi(chainMessage ChainMessage) bool {
+func IsHangingApi(chainMessage ChainMessageForSend) bool {
 	return chainMessage.GetApi().Category.HangingApi
 }
 
-func GetComputeUnits(chainMessage ChainMessage) uint64 {
+func GetComputeUnits(chainMessage ChainMessageForSend) uint64 {
 	return chainMessage.GetApi().ComputeUnits
 }
 
-func GetStateful(chainMessage ChainMessage) uint32 {
+func GetStateful(chainMessage ChainMessageForSend) uint32 {
 	return chainMessage.GetApi().Category.Stateful
 }

protocol/chainlib/chainlib.go

@@ -85,6 +85,7 @@ type ChainMessage interface {
 }
 
 type ChainMessageForSend interface {
+	TimeoutOverride(...time.Duration) time.Duration
 	GetApi() *spectypes.Api
 	GetRPCMessage() rpcInterfaceMessages.GenericMessage
 	GetApiCollection() *spectypes.ApiCollection

protocol/chainlib/common.go

@@ -1,6 +1,7 @@
 package chainlib
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net"
@@ -86,6 +87,13 @@ func (bcp *BaseChainProxy) GetChainProxyInformation() (common.NodeUrl, string) {
 	return bcp.NodeUrl, bcp.ChainID
 }
 
+func (bcp *BaseChainProxy) CapTimeoutForSend(ctx context.Context, chainMessage ChainMessageForSend) (context.Context, context.CancelFunc) {
+	relayTimeout := GetRelayTimeout(chainMessage, bcp.averageBlockTime)
+	processingTimeout := common.GetTimeoutForProcessing(relayTimeout, GetTimeoutInfo(chainMessage))
+	connectCtx, cancel := bcp.NodeUrl.LowerContextTimeout(ctx, processingTimeout)
+	return connectCtx, cancel
+}
+
 func extractDappIDFromFiberContext(c *fiber.Ctx) (dappID string) {
 	// Read the dappID from the headers
 	dappID = c.Get("dapp-id")
@@ -299,21 +307,21 @@ func CompareRequestedBlockInBatch(firstRequestedBlock int64, second int64) (late
 	return returnBigger(firstRequestedBlock, second)
 }
 
-func GetRelayTimeout(chainMessage ChainMessage, chainParser ChainParser, timeouts int) time.Duration {
+func GetRelayTimeout(chainMessage ChainMessageForSend, averageBlockTime time.Duration) time.Duration {
 	if chainMessage.TimeoutOverride() != 0 {
 		return chainMessage.TimeoutOverride()
 	}
 	// Calculate extra RelayTimeout
 	extraRelayTimeout := time.Duration(0)
 	if IsHangingApi(chainMessage) {
-		_, extraRelayTimeout, _, _ = chainParser.ChainBlockStats()
+		extraRelayTimeout = averageBlockTime * 2
 	}
 	relayTimeAddition := common.GetTimePerCu(GetComputeUnits(chainMessage))
 	if chainMessage.GetApi().TimeoutMs > 0 {
 		relayTimeAddition = time.Millisecond * time.Duration(chainMessage.GetApi().TimeoutMs)
 	}
 	// Set relay timout, increase it every time we fail a relay on timeout
-	return extraRelayTimeout + time.Duration(timeouts+1)*relayTimeAddition + common.AverageWorldLatency
+	return extraRelayTimeout + relayTimeAddition + common.AverageWorldLatency
 }
 
 // setup a common preflight and cors configuration allowing wild cards and preflight caching.
@@ -416,3 +424,11 @@ func (rd *RefererData) SendReferer(refererMatchString string, chainId string, ms
 	rd.ReferrerClient.AppendReferrer(metrics.NewReferrerRequest(refererMatchString, chainId, msg, referer, origin, userAgent))
 	return nil
 }
+
+func GetTimeoutInfo(chainMessage ChainMessageForSend) common.TimeoutInfo {
+	return common.TimeoutInfo{
+		CU:       chainMessage.GetApi().ComputeUnits,
+		Hanging:  IsHangingApi(chainMessage),
+		Stateful: GetStateful(chainMessage),
+	}
+}

protocol/chainlib/common_test_utils.go

@@ -86,19 +86,19 @@ func generateCombinations(arr []string) [][]string {
 
 // generates a chain parser, a chain fetcher messages based on it
 // apiInterface can either be an ApiInterface string as in spectypes.ApiInterfaceXXX or a number for an index in the apiCollections
-func CreateChainLibMocks(ctx context.Context, specIndex string, apiInterface string, serverCallback http.HandlerFunc, getToTopMostPath string, services []string) (cpar ChainParser, crout ChainRouter, cfetc chaintracker.ChainFetcher, closeServer func(), errRet error) {
+func CreateChainLibMocks(ctx context.Context, specIndex string, apiInterface string, serverCallback http.HandlerFunc, getToTopMostPath string, services []string) (cpar ChainParser, crout ChainRouter, cfetc chaintracker.ChainFetcher, closeServer func(), endpointRet *lavasession.RPCProviderEndpoint, errRet error) {
 	closeServer = nil
 	spec, err := keepertest.GetASpec(specIndex, getToTopMostPath, nil, nil)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 	index, err := strconv.Atoi(apiInterface)
 	if err == nil && index < len(spec.ApiCollections) {
 		apiInterface = spec.ApiCollections[index].CollectionData.ApiInterface
 	}
 	chainParser, err := NewChainParser(apiInterface)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 	var chainRouter ChainRouter
 	chainParser.SetSpec(spec)
@@ -111,15 +111,15 @@ func CreateChainLibMocks(ctx context.Context, specIndex string, apiInterface str
 	}
 	addons, extensions, err := chainParser.SeparateAddonsExtensions(services)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 
 	if apiInterface == spectypes.APIInterfaceGrpc {
 		// Start a new gRPC server using the buffered connection
 		grpcServer := grpc.NewServer()
-		lis, err := net.Listen("tcp", "localhost:0")
+		lis, err := net.Listen("tcp", "127.0.0.1:0")
 		if err != nil {
-			return nil, nil, nil, closeServer, err
+			return nil, nil, nil, closeServer, nil, err
 		}
 		endpoint.NodeUrls = append(endpoint.NodeUrls, common.NodeUrl{Url: lis.Addr().String(), Addons: addons})
 		allCombinations := generateCombinations(extensions)
@@ -138,19 +138,19 @@ func CreateChainLibMocks(ctx context.Context, specIndex string, apiInterface str
 		time.Sleep(10 * time.Millisecond)
 		chainRouter, err = GetChainRouter(ctx, 1, endpoint, chainParser)
 		if err != nil {
-			return nil, nil, nil, closeServer, err
+			return nil, nil, nil, closeServer, nil, err
 		}
 	} else {
 		mockServer := httptest.NewServer(serverCallback)
 		closeServer = mockServer.Close
 		endpoint.NodeUrls = append(endpoint.NodeUrls, common.NodeUrl{Url: mockServer.URL, Addons: addons})
 		chainRouter, err = GetChainRouter(ctx, 1, endpoint, chainParser)
 		if err != nil {
-			return nil, nil, nil, closeServer, err
+			return nil, nil, nil, closeServer, nil, err
 		}
 	}
 	chainFetcher := NewChainFetcher(ctx, &ChainFetcherOptions{chainRouter, chainParser, endpoint, nil})
-	return chainParser, chainRouter, chainFetcher, closeServer, err
+	return chainParser, chainRouter, chainFetcher, closeServer, endpoint, err
 }
 
 type TestStruct struct {

protocol/chainlib/grpc.go

@@ -310,7 +310,7 @@ func (apil *GrpcChainListener) Serve(ctx context.Context, cmdFlags common.Consum
 		grpcHeaders := convertToMetadataMapOfSlices(metadataValues)
 		utils.LavaFormatDebug("in <<< GRPC Relay ",
 			utils.LogAttr("GUID", ctx),
-			utils.LogAttr("method", method),
+			utils.LogAttr("_method", method),
 			utils.LogAttr("headers", grpcHeaders),
 		)
 		metricsData := metrics.NewRelayAnalytics(dappID, apil.endpoint.ChainID, apiInterface)
@@ -520,14 +520,14 @@ func (cp *GrpcChainProxy) SendNodeMsg(ctx context.Context, ch chan interface{},
 	}
 	if debug {
 		utils.LavaFormatDebug("provider sending node message",
-			utils.Attribute{Key: "method", Value: nodeMessage.Path},
+			utils.Attribute{Key: "_method", Value: nodeMessage.Path},
 			utils.Attribute{Key: "headers", Value: metadataMap},
 			utils.Attribute{Key: "apiInterface", Value: "grpc"},
 		)
 	}
 	var respHeaders metadata.MD
 	response := msgFactory.NewMessage(methodDescriptor.GetOutputType())
-	connectCtx, cancel := cp.NodeUrl.LowerContextTimeout(ctx, chainMessage, cp.averageBlockTime)
+	connectCtx, cancel := cp.CapTimeoutForSend(ctx, chainMessage)
 	defer cancel()
 	err = conn.Invoke(connectCtx, "/"+nodeMessage.Path, msg, response, grpc.Header(&respHeaders))
 	if err != nil {

protocol/chainlib/grpc_test.go

@@ -142,7 +142,7 @@ func TestGrpcChainProxy(t *testing.T) {
 		// Handle the incoming request and provide the desired response
 		wasCalled = true
 	})
-	chainParser, chainProxy, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandle, "../../", nil)
+	chainParser, chainProxy, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandle, "../../", nil)
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainProxy)
@@ -169,7 +169,7 @@ func TestParsingRequestedBlocksHeadersGrpc(t *testing.T) {
 			w.WriteHeader(244591)
 		}
 	})
-	chainParser, chainRouter, _, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandler, "../../", nil)
+	chainParser, chainRouter, _, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandler, "../../", nil)
 	require.NoError(t, err)
 	defer func() {
 		if closeServer != nil {
@@ -237,7 +237,7 @@ func TestSettingBlocksHeadersGrpc(t *testing.T) {
 			w.WriteHeader(244591)
 		}
 	})
-	chainParser, chainRouter, _, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandler, "../../", nil)
+	chainParser, chainRouter, _, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceGrpc, serverHandler, "../../", nil)
 	require.NoError(t, err)
 	defer func() {
 		if closeServer != nil {

protocol/chainlib/jsonRPC.go

@@ -446,7 +446,7 @@ func (apil *JsonRPCChainListener) Serve(ctx context.Context, cmdFlags common.Con
 		utils.LavaFormatDebug("in <<<",
 			utils.LogAttr("GUID", ctx),
 			utils.LogAttr("seed", msgSeed),
-			utils.LogAttr("msg", logFormattedMsg),
+			utils.LogAttr("_msg", logFormattedMsg),
 			utils.LogAttr("dappID", dappID),
 			utils.LogAttr("headers", headers),
 		)
@@ -596,7 +596,7 @@ func (cp *JrpcChainProxy) sendBatchMessage(ctx context.Context, nodeMessage *rpc
 		}
 	}
 	// set context with timeout
-	connectCtx, cancel := cp.NodeUrl.LowerContextTimeout(ctx, chainMessage, cp.averageBlockTime)
+	connectCtx, cancel := cp.CapTimeoutForSend(ctx, chainMessage)
 	defer cancel()
 
 	cp.NodeUrl.SetIpForwardingIfNecessary(ctx, rpc.SetHeader)
@@ -668,7 +668,7 @@ func (cp *JrpcChainProxy) SendNodeMsg(ctx context.Context, ch chan interface{},
 	} else {
 		// we use the minimum timeout between the two, spec or context. to prevent the provider from hanging
 		// we don't use the context alone so the provider won't be hanging forever by an attack
-		connectCtx, cancel := cp.NodeUrl.LowerContextTimeout(ctx, chainMessage, cp.averageBlockTime)
+		connectCtx, cancel := cp.CapTimeoutForSend(ctx, chainMessage)
 		defer cancel()
 
 		cp.NodeUrl.SetIpForwardingIfNecessary(ctx, rpc.SetHeader)

protocol/chainlib/jsonRPC_test.go

@@ -140,7 +140,7 @@ func TestJsonRpcChainProxy(t *testing.T) {
 		fmt.Fprint(w, `{"jsonrpc":"2.0","id":1,"result":"0x10a7a08"}`)
 	})
 
-	chainParser, chainProxy, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
+	chainParser, chainProxy, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainProxy)
@@ -164,7 +164,7 @@ func TestAddonAndVerifications(t *testing.T) {
 		fmt.Fprint(w, `{"jsonrpc":"2.0","id":1,"result":"0xf9ccdff90234a064"}`)
 	})
 
-	chainParser, chainRouter, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", []string{"debug"})
+	chainParser, chainRouter, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", []string{"debug"})
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainRouter)
@@ -197,7 +197,7 @@ func TestExtensions(t *testing.T) {
 	})
 
 	specname := "ETH1"
-	chainParser, chainRouter, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, specname, spectypes.APIInterfaceJsonRPC, serverHandle, "../../", []string{"archive"})
+	chainParser, chainRouter, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, specname, spectypes.APIInterfaceJsonRPC, serverHandle, "../../", []string{"archive"})
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainRouter)
@@ -279,7 +279,7 @@ func TestJsonRpcBatchCall(t *testing.T) {
 		fmt.Fprint(w, response)
 	})
 
-	chainParser, chainProxy, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
+	chainParser, chainProxy, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainProxy)
@@ -320,7 +320,7 @@ func TestJsonRpcBatchCallSameID(t *testing.T) {
 		fmt.Fprint(w, response)
 	})
 
-	chainParser, chainProxy, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
+	chainParser, chainProxy, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "ETH1", spectypes.APIInterfaceJsonRPC, serverHandle, "../../", nil)
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainProxy)

protocol/chainlib/rest.go

@@ -300,7 +300,7 @@ func (apil *RestChainListener) Serve(ctx context.Context, cmdFlags common.Consum
 		analytics := metrics.NewRelayAnalytics(dappID, chainID, apiInterface)
 		utils.LavaFormatDebug("in <<<",
 			utils.LogAttr("GUID", ctx),
-			utils.LogAttr("path", path),
+			utils.LogAttr("_path", path),
 			utils.LogAttr("dappID", dappID),
 			utils.LogAttr("msgSeed", msgSeed),
 			utils.LogAttr("headers", restHeaders),
@@ -366,7 +366,7 @@ func (apil *RestChainListener) Serve(ctx context.Context, cmdFlags common.Consum
 		defer cancel() // incase there's a problem make sure to cancel the connection
 		utils.LavaFormatDebug("in <<<",
 			utils.LogAttr("GUID", ctx),
-			utils.LogAttr("path", path),
+			utils.LogAttr("_path", path),
 			utils.LogAttr("dappID", dappID),
 			utils.LogAttr("msgSeed", msgSeed),
 			utils.LogAttr("headers", restHeaders),
@@ -473,7 +473,7 @@ func (rcp *RestChainProxy) SendNodeMsg(ctx context.Context, ch chan interface{},
 	urlPath := rcp.NodeUrl.Url + nodeMessage.Path
 
 	// set context with timeout
-	connectCtx, cancel := rcp.NodeUrl.LowerContextTimeout(ctx, chainMessage, rcp.averageBlockTime)
+	connectCtx, cancel := rcp.CapTimeoutForSend(ctx, chainMessage)
 	defer cancel()
 
 	req, err := http.NewRequestWithContext(connectCtx, connectionTypeSlected, rcp.NodeUrl.AuthConfig.AddAuthPath(urlPath), msgBuffer)
@@ -496,7 +496,7 @@ func (rcp *RestChainProxy) SendNodeMsg(ctx context.Context, ch chan interface{},
 
 	if debug {
 		utils.LavaFormatDebug("provider sending node message",
-			utils.Attribute{Key: "method", Value: nodeMessage.Path},
+			utils.Attribute{Key: "_method", Value: nodeMessage.Path},
 			utils.Attribute{Key: "headers", Value: req.Header},
 			utils.Attribute{Key: "apiInterface", Value: "rest"},
 		)

protocol/chainlib/rest_test.go

@@ -135,7 +135,7 @@ func TestRestChainProxy(t *testing.T) {
 		w.WriteHeader(http.StatusOK)
 		fmt.Fprint(w, `{"block": { "header": {"height": "244591"}}}`)
 	})
-	chainParser, chainProxy, chainFetcher, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
+	chainParser, chainProxy, chainFetcher, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
 	require.NoError(t, err)
 	require.NotNil(t, chainParser)
 	require.NotNil(t, chainProxy)
@@ -166,7 +166,7 @@ func TestParsingRequestedBlocksHeadersRest(t *testing.T) {
 			fmt.Fprint(w, `{"block": { "header": {"height": "244591"}}}`)
 		}
 	})
-	chainParser, chainRouter, _, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
+	chainParser, chainRouter, _, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
 	require.NoError(t, err)
 	defer func() {
 		if closeServer != nil {
@@ -236,7 +236,7 @@ func TestSettingRequestedBlocksHeadersRest(t *testing.T) {
 		}
 		fmt.Fprint(w, `{"block": { "header": {"height": "244591"}}}`)
 	})
-	chainParser, chainRouter, _, closeServer, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
+	chainParser, chainRouter, _, closeServer, _, err := CreateChainLibMocks(ctx, "LAV1", spectypes.APIInterfaceRest, serverHandler, "../../", nil)
 	require.NoError(t, err)
 	defer func() {
 		if closeServer != nil {