chore: PRT - fireblocks with UTXO support

protocol/chainlib/base_chain_parser.go

@@ -20,7 +20,11 @@ import (
 	spectypes "github.com/lavanet/lava/v5/x/spec/types"
 )
 
-var AllowMissingApisByDefault = true
+var (
+	SkipPolicyVerification    = false
+	AllowMissingApisByDefault = true
+	SkipWebsocketVerification = false
+)
 
 type PolicyInf interface {
 	GetSupportedAddons(specID string) (addons []string, err error)
@@ -98,12 +102,15 @@ func (bcp *BaseChainParser) isAddon(addon string) bool {
 }
 
 func (bcp *BaseChainParser) isExtension(extension string) bool {
-	return bcp.extensionParser.AllowedExtension(extension)
+	return bcp.extensionParser.AllowedExtension(extension, SkipPolicyVerification)
 }
 
 // use while bcp locked.
 func (bcp *BaseChainParser) validateAddons(nodeMessage *baseChainMessageContainer) error {
 	var addon string
+	if SkipPolicyVerification {
+		return nil
+	}
 	if addon = GetAddon(nodeMessage); addon != "" { // check we have an addon
 		if allowed := bcp.allowedAddons[addon]; !allowed { // check addon is allowed
 			return utils.LavaFormatError("consumer policy does not allow addon", nil,

protocol/chainlib/chain_fetcher.go

@@ -183,7 +183,7 @@ func getExtensionsForVerification(verification VerificationContainer, chainParse
 		ConnectionType: verification.ConnectionType,
 	}
 
-	if chainParser.IsTagInCollection(spectypes.FUNCTION_TAG_SUBSCRIBE, collectionKey) {
+	if chainParser.IsTagInCollection(spectypes.FUNCTION_TAG_SUBSCRIBE, collectionKey) && !SkipWebsocketVerification {
 		if verification.Extension == "" {
 			extensions = []string{WebSocketExtension}
 		} else {

protocol/chainlib/chain_router.go

@@ -322,7 +322,7 @@ func newChainRouter(ctx context.Context, nConns uint, rpcProviderEndpoint lavase
 			break
 		}
 	}
-	if !IgnoreWsEnforcementForTestCommands && hasSubscriptionInSpec && apiCollection.Enabled && !webSocketSupported {
+	if !IgnoreWsEnforcementForTestCommands && hasSubscriptionInSpec && apiCollection.Enabled && !webSocketSupported && !SkipWebsocketVerification {
 		return nil, utils.LavaFormatError("subscriptions are applicable for this chain, but websocket is not provided in 'supported' map. By not setting ws/wss your provider wont be able to accept ws subscriptions, therefore might receive less rewards and lower QOS score.", nil,
 			utils.LogAttr("apiInterface", apiCollection.CollectionData.ApiInterface),
 			utils.LogAttr("supportedMap", supportedMap),

protocol/chainlib/extensionslib/extension_parser.go

@@ -55,8 +55,8 @@ func (ep *ExtensionParser) GetConfiguredExtensions() map[ExtensionKey]*spectypes
 	return ep.configuredExtensions
 }
 
-func (ep *ExtensionParser) AllowedExtension(extension string) bool {
-	if extension == "" {
+func (ep *ExtensionParser) AllowedExtension(extension string, skip bool) bool {
+	if extension == "" || skip {
 		return true
 	}
 	_, ok := ep.allowedExtensions[extension]

protocol/common/cobra_common.go

@@ -53,8 +53,12 @@ const (
 	LimitParallelWebsocketConnectionsPerIpFlag   = "limit-parallel-websocket-connections-per-ip"
 	LimitWebsocketIdleTimeFlag                   = "limit-websocket-connection-idle-time"
 	RateLimitRequestPerSecondFlag                = "rate-limit-requests-per-second"
+	SkipPolicyVerificationFlag                   = "skip-policy-verification"
+	SkipWebsocketVerificationFlag                = "skip-websocket-verification"
 	// specification default flags
-	AllowMissingApisByDefaultFlagName = "allow-missing-apis-by-default"
+	AllowMissingApisByDefaultFlagName      = "allow-missing-apis-by-default"
+	PeriodicProbeProvidersFlagName         = "enable-periodic-probe-providers"
+	PeriodicProbeProvidersIntervalFlagName = "periodic-probe-providers-interval"
 
 	ShowProviderEndpointInMetricsFlagName = "show-provider-address-in-metrics"
 )

protocol/lavasession/consumer_session_manager.go

@@ -27,15 +27,18 @@ const (
 )
 
 var (
-	retrySecondChanceAfter = time.Minute * 3
-	DebugProbes            = false
+	retrySecondChanceAfter         = time.Minute * 3
+	DebugProbes                    = false
+	PeriodicProbeProviders         = false
+	PeriodicProbeProvidersInterval = 5 * time.Second
 )
 
 // created with NewConsumerSessionManager
 type ConsumerSessionManager struct {
 	rpcEndpoint    *RPCEndpoint // used to filter out endpoints
 	lock           sync.RWMutex
 	pairing        map[string]*ConsumerSessionsWithProvider // key == provider address
+	rawPairing     map[uint64]*ConsumerSessionsWithProvider // key == provider index in pairing. Used for periodic probing of providers
 	currentEpoch   uint64
 	numberOfResets uint64
 
@@ -88,6 +91,8 @@ func (csm *ConsumerSessionManager) UpdateAllProviders(epoch uint64, pairingList
 	csm.lock.Lock()         // start by locking the class lock.
 	defer csm.lock.Unlock() // we defer here so in case we return an error it will unlock automatically.
 
+	csm.rawPairing = pairingList
+
 	if epoch <= csm.atomicReadCurrentEpoch() { // sentry shouldn't update an old epoch or current epoch
 		return utils.LavaFormatError("trying to update provider list for older epoch", nil, utils.Attribute{Key: "epoch", Value: epoch}, utils.Attribute{Key: "currentEpoch", Value: csm.atomicReadCurrentEpoch()})
 	}
@@ -102,6 +107,9 @@ func (csm *ConsumerSessionManager) UpdateAllProviders(epoch uint64, pairingList
 	csm.reportedProviders.Reset()
 	csm.pairingAddressesLength = uint64(pairingListLength)
 	csm.numberOfResets = 0
+
+	providerAddressToEndpoint := map[string]string{}
+
 	csm.RemoveAddonAddresses("", nil)
 	// Reset the pairingPurge.
 	// This happens only after an entire epoch. so its impossible to have session connected to the old purged list
@@ -111,11 +119,13 @@ func (csm *ConsumerSessionManager) UpdateAllProviders(epoch uint64, pairingList
 	for idx, provider := range pairingList {
 		csm.pairingAddresses[idx] = provider.PublicLavaAddress
 		csm.pairing[provider.PublicLavaAddress] = provider
+		providerAddressToEndpoint[provider.PublicLavaAddress] = provider.Endpoints[0].NetworkAddress
 	}
 	csm.setValidAddressesToDefaultValue("", nil) // the starting point is that valid addresses are equal to pairing addresses.
 	// reset session related metrics
 	go csm.consumerMetricsManager.ResetSessionRelatedMetrics()
 	go csm.providerOptimizer.UpdateWeights(CalcWeightsByStake(pairingList), epoch)
+	go csm.consumerMetricsManager.ResetBlockedProvidersMetrics(csm.rpcEndpoint.ChainID, csm.rpcEndpoint.ApiInterface, providerAddressToEndpoint)
 
 	utils.LavaFormatDebug("updated providers", utils.Attribute{Key: "epoch", Value: epoch}, utils.Attribute{Key: "spec", Value: csm.rpcEndpoint.Key()})
 	return nil
@@ -191,6 +201,21 @@ func (csm *ConsumerSessionManager) closePurgedUnusedPairingsConnections() {
 	}
 }
 
+func (csm *ConsumerSessionManager) PeriodicProbeProviders(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+
+	for {
+		select {
+		case <-ticker.C:
+			if csm.rawPairing != nil {
+				csm.probeProviders(ctx, csm.rawPairing, csm.atomicReadCurrentEpoch())
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 func (csm *ConsumerSessionManager) probeProviders(ctx context.Context, pairingList map[uint64]*ConsumerSessionsWithProvider, epoch uint64) error {
 	guid := utils.GenerateUniqueIdentifier()
 	ctx = utils.AppendUniqueIdentifier(ctx, guid)
@@ -207,6 +232,7 @@ func (csm *ConsumerSessionManager) probeProviders(ctx context.Context, pairingLi
 			defer wg.Done()
 			latency, providerAddress, err := csm.probeProvider(ctx, consumerSessionsWithProvider, epoch, false)
 			success := err == nil // if failure then regard it in availability
+			csm.consumerMetricsManager.SetProviderLiveness(csm.rpcEndpoint.ChainID, providerAddress, consumerSessionWithProvider.Endpoints[0].NetworkAddress, success)
 			csm.providerOptimizer.AppendProbeRelayData(providerAddress, latency, success)
 		}(consumerSessionWithProvider)
 	}
@@ -830,6 +856,13 @@ func (csm *ConsumerSessionManager) removeAddressFromValidAddresses(address strin
 			csm.currentlyBlockedProviderAddresses = append(csm.currentlyBlockedProviderAddresses, address)
 			// sort the blocked provider list by cu served
 			csm.sortBlockedProviderListByCuServed()
+			provider, ok := csm.pairing[addr]
+			if ok {
+				info := csm.RPCEndpoint()
+				go func(networkAddress string, chainId string, apiInterface string, providerAddress string) {
+					csm.consumerMetricsManager.SetBlockedProvider(chainId, apiInterface, providerAddress, networkAddress, true)
+				}(provider.Endpoints[0].NetworkAddress, info.ChainID, info.ApiInterface, addr)
+			}
 			return nil
 		}
 	}
@@ -1005,7 +1038,11 @@ func (csm *ConsumerSessionManager) validateAndReturnBlockedProviderToValidAddres
 			csm.RemoveAddonAddresses("", nil)
 			// Reset redemption status
 			if provider, ok := csm.pairing[providerAddress]; ok {
+				info := csm.RPCEndpoint()
 				provider.atomicWriteBlockedStatus(BlockedProviderSessionUnusedStatus)
+				go func(networkAddress string, chainId string, apiInterface string, providerAddress string) {
+					csm.consumerMetricsManager.SetBlockedProvider(chainId, apiInterface, providerAddress, networkAddress, false)
+				}(provider.Endpoints[0].NetworkAddress, info.ChainID, info.ApiInterface, providerAddress)
 			}
 			return
 		}

protocol/lavasession/consumer_session_manager_test.go

@@ -231,7 +231,10 @@ func (bpm DirectiveHeaders) GetBlockedProviders() []string {
 	}
 	blockedProviders, ok := bpm.directiveHeaders[common.BLOCK_PROVIDERS_ADDRESSES_HEADER_NAME]
 	if ok {
-		return strings.Split(blockedProviders, ",")
+		blockProviders := strings.Split(blockedProviders, ",")
+		if len(blockProviders) <= 2 {
+			return blockProviders
+		}
 	}
 	return nil
 }

protocol/metrics/consumer_metrics_manager.go

@@ -71,6 +71,8 @@ type ConsumerMetricsManager struct {
 	relayProcessingLatencyAfterProvider         *prometheus.GaugeVec
 	averageProcessingLatency                    map[string]*LatencyTracker
 	consumerOptimizerQoSClient                  *ConsumerOptimizerQoSClient
+	providerLivenessMetric                      *prometheus.GaugeVec
+	blockedProviderMetric                       *MappedLabelsGaugeVec
 }
 
 type ConsumerMetricsManagerOptions struct {
@@ -153,6 +155,22 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 		Help: "The latency of requests requested by the consumer over time.",
 	}, []string{"spec", "apiInterface"})
 
+	providerLivenessMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "lava_consumer_provider_liveness",
+		Help: "The liveness of connected provider based on probe",
+	}, []string{"spec", "provider_address", "provider_endpoint"})
+
+	blockedProviderMetricLabels := []string{"spec", "apiInterface", "provider_address"}
+	if ShowProviderEndpointInMetrics {
+		blockedProviderMetricLabels = append(blockedProviderMetricLabels, "provider_endpoint")
+	}
+
+	blockedProviderMetric := NewMappedLabelsGaugeVec(MappedLabelsMetricOpts{
+		Name:   "lava_consumer_provider_blocked",
+		Help:   "Is provider blocked. 1-blocked, 0-not blocked",
+		Labels: blockedProviderMetricLabels,
+	})
+
 	qosMetricLabels := []string{"spec", "apiInterface", "provider_address", "qos_metric"}
 	if ShowProviderEndpointInMetrics {
 		qosMetricLabels = append(qosMetricLabels, "provider_endpoint")
@@ -255,6 +273,7 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 	prometheus.MustRegister(totalErroredMetric)
 	prometheus.MustRegister(blockMetric)
 	prometheus.MustRegister(latencyMetric)
+	prometheus.MustRegister(providerLivenessMetric)
 	prometheus.MustRegister(latestProviderRelay)
 	prometheus.MustRegister(virtualEpochMetric)
 	prometheus.MustRegister(endpointsHealthChecksOkMetric)
@@ -313,6 +332,8 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 		totalLoLSuccessMetric:                       totalLoLSuccessMetric,
 		totalLoLErrorsMetric:                        totalLoLErrorsMetric,
 		consumerOptimizerQoSClient:                  options.ConsumerOptimizerQoSClient,
+		providerLivenessMetric:                      providerLivenessMetric,
+		blockedProviderMetric:                       blockedProviderMetric,
 	}
 
 	http.Handle("/metrics", promhttp.Handler())
@@ -569,6 +590,19 @@ func (pme *ConsumerMetricsManager) ResetSessionRelatedMetrics() {
 	pme.providerRelays = map[string]uint64{}
 }
 
+func (pme *ConsumerMetricsManager) ResetBlockedProvidersMetrics(chainId, apiInterface string, providers map[string]string) {
+	if pme == nil {
+		return
+	}
+	pme.lock.Lock()
+	defer pme.lock.Unlock()
+	pme.blockedProviderMetric.Reset()
+	for provider, endpoint := range providers {
+		labels := map[string]string{"spec": chainId, "apiInterface": apiInterface, "provider_address": provider, "provider_endpoint": endpoint}
+		pme.blockedProviderMetric.WithLabelValues(labels).Set(0)
+	}
+}
+
 func (pme *ConsumerMetricsManager) SetVersion(version string) {
 	if pme == nil {
 		return
@@ -627,6 +661,33 @@ func (pme *ConsumerMetricsManager) SetLoLResponse(success bool) {
 	}
 }
 
+func (pme *ConsumerMetricsManager) SetProviderLiveness(chainId string, providerAddress string, providerEndpoint string, isAlive bool) {
+	if pme == nil {
+		return
+	}
+
+	var value float64 = 0
+	if isAlive {
+		value = 1
+	}
+
+	pme.providerLivenessMetric.WithLabelValues(chainId, providerAddress, providerEndpoint).Set(value)
+}
+
+func (pme *ConsumerMetricsManager) SetBlockedProvider(chainId, apiInterface, providerAddress, providerEndpoint string, isBlocked bool) {
+	if pme == nil {
+		return
+	}
+	var value float64 = 0
+	if isBlocked {
+		value = 1
+	}
+	labels := map[string]string{"spec": chainId, "apiInterface": apiInterface, "provider_address": providerAddress, "provider_endpoint": providerEndpoint}
+	pme.lock.Lock()
+	defer pme.lock.Unlock()
+	pme.blockedProviderMetric.WithLabelValues(labels).Set(value)
+}
+
 func (pme *ConsumerMetricsManager) handleOptimizerQoS(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)

protocol/rpcconsumer/rpcconsumer.go

@@ -413,6 +413,9 @@ func (rpcc *RPCConsumer) CreateConsumerEndpoint(
 	// Create active subscription provider storage for each unique chain
 	activeSubscriptionProvidersStorage := lavasession.NewActiveSubscriptionProvidersStorage()
 	consumerSessionManager := lavasession.NewConsumerSessionManager(rpcEndpoint, optimizer, consumerMetricsManager, consumerReportsManager, consumerAddr.String(), activeSubscriptionProvidersStorage)
+	if lavasession.PeriodicProbeProviders {
+		go consumerSessionManager.PeriodicProbeProviders(ctx, lavasession.PeriodicProbeProvidersInterval)
+	}
 	// Register For Updates
 	rpcc.consumerStateTracker.RegisterConsumerSessionManagerForPairingUpdates(ctx, consumerSessionManager, options.staticProvidersList)
 
@@ -782,10 +785,15 @@ rpcconsumer consumer_examples/full_consumer_example.yml --cache-be "127.0.0.1:77
 	cmdRPCConsumer.Flags().Int64Var(&chainlib.MaximumNumberOfParallelWebsocketConnectionsPerIp, common.LimitParallelWebsocketConnectionsPerIpFlag, chainlib.MaximumNumberOfParallelWebsocketConnectionsPerIp, "limit number of parallel connections to websocket, per ip, default is unlimited (0)")
 	cmdRPCConsumer.Flags().Int64Var(&chainlib.MaxIdleTimeInSeconds, common.LimitWebsocketIdleTimeFlag, chainlib.MaxIdleTimeInSeconds, "limit the idle time in seconds for a websocket connection, default is 20 minutes ( 20 * 60 )")
 	cmdRPCConsumer.Flags().DurationVar(&chainlib.WebSocketBanDuration, common.BanDurationForWebsocketRateLimitExceededFlag, chainlib.WebSocketBanDuration, "once websocket rate limit is reached, user will be banned Xfor a duration, default no ban")
+	cmdRPCConsumer.Flags().BoolVar(&chainlib.SkipPolicyVerification, common.SkipPolicyVerificationFlag, chainlib.SkipPolicyVerification, "skip policy verifications, this flag will skip onchain policy verification and will use the static provider list")
+
 	// lava over lava backup
 	cmdRPCConsumer.Flags().Bool(LavaOverLavaBackupFlagName, true, "enable lava over lava backup to regular rpc calls")
 	// allow missing apis by default
 	cmdRPCConsumer.Flags().BoolVar(&chainlib.AllowMissingApisByDefault, common.AllowMissingApisByDefaultFlagName, true, "allows missing apis to be proxied to the provider by default, set flase to block missing apis in the spec")
+	cmdRPCConsumer.Flags().BoolVar(&lavasession.PeriodicProbeProviders, common.PeriodicProbeProvidersFlagName, lavasession.PeriodicProbeProviders, "enable periodic probing of providers")
+	cmdRPCConsumer.Flags().DurationVar(&lavasession.PeriodicProbeProvidersInterval, common.PeriodicProbeProvidersIntervalFlagName, lavasession.PeriodicProbeProvidersInterval, "interval for periodic probing of providers")
+
 	common.AddRollingLogConfig(cmdRPCConsumer)
 	return cmdRPCConsumer
 }

protocol/rpcconsumer/rpcconsumer_server.go

@@ -319,7 +319,7 @@ func (rpccs *RPCConsumerServer) getLatestBlock() uint64 {
 	if numProviders > 0 && latestKnownBlock > 0 {
 		return uint64(latestKnownBlock)
 	}
-	utils.LavaFormatWarning("no information on latest block", nil, utils.Attribute{Key: "latest block", Value: 0})
+	utils.LavaFormatDebug("no information on latest block", utils.Attribute{Key: "latest block", Value: 0})
 	return 0
 }
 

protocol/rpcprovider/rpcprovider.go

@@ -881,6 +881,7 @@ rpcprovider 127.0.0.1:3333 OSMOSIS tendermintrpc "wss://www.node-path.com:80,htt
 	cmdRPCProvider.Flags().String(common.UseStaticSpecFlag, "", "load offline spec provided path to spec file, used to test specs before they are proposed on chain, example for spec with inheritance: --use-static-spec ./specs/mainnet-1/specs/ibc.json,./specs/mainnet-1/specs/tendermint.json,./specs/mainnet-1/specs/cosmossdk.json,./specs/mainnet-1/specs/ethermint.json,./specs/mainnet-1/specs/ethereum.json,./specs/mainnet-1/specs/evmos.json")
 	cmdRPCProvider.Flags().Uint64(common.RateLimitRequestPerSecondFlag, 0, "Measuring the load relative to this number for feedback - per second - per chain - default unlimited. Given Y simultaneous relay calls, a value of X  and will measure Y/X load rate.")
 	cmdRPCProvider.Flags().BoolVar(&chainlib.AllowMissingApisByDefault, common.AllowMissingApisByDefaultFlagName, true, "allows missing apis to be proxied to the node by default, set false to block missing apis in the spec, might result in degraded performance if spec is misconfigured")
+	cmdRPCProvider.Flags().BoolVar(&chainlib.SkipWebsocketVerification, common.SkipWebsocketVerificationFlag, false, "skip websocket verification")
 	cmdRPCProvider.Flags().BoolVar(&metrics.ShowProviderEndpointInProviderMetrics, common.ShowProviderEndpointInMetricsFlagName, metrics.ShowProviderEndpointInProviderMetrics, "show provider endpoint in provider metrics")
 	common.AddRollingLogConfig(cmdRPCProvider)
 	return cmdRPCProvider

protocol/upgrade/protocol_version.go

@@ -26,7 +26,7 @@ func GetCurrentVersion() ProtocolVersion {
 func (pv *ProtocolVersion) ValidateProtocolVersion(incoming *updaters.ProtocolVersionResponse) error {
 	// check min version
 	if HasVersionMismatch(incoming.Version.ConsumerMin, lavaProtocolVersion.ConsumerVersion) || HasVersionMismatch(incoming.Version.ProviderMin, lavaProtocolVersion.ProviderVersion) {
-		utils.LavaFormatFatal("minimum protocol version mismatch!, you must update your protocol version to at least the minimum required protocol version",
+		return utils.LavaFormatError("minimum protocol version mismatch!, you must update your protocol version to at least the minimum required protocol version",
 			nil,
 			utils.Attribute{Key: "required (on-chain) consumer minimum version:", Value: incoming.Version.ConsumerMin},
 			utils.Attribute{Key: "required (on-chain) provider minimum version", Value: incoming.Version.ProviderMin},