chore: PRT - Add offline policy v4.2.1 - DO NOT MERGE

protocol/chainlib/base_chain_parser.go

@@ -20,7 +20,10 @@ import (
 	spectypes "github.com/lavanet/lava/v4/x/spec/types"
 )
 
-var AllowMissingApisByDefault = true
+var (
+	SkipPolicyVerification    = false
+	AllowMissingApisByDefault = true
+)
 
 type PolicyInf interface {
 	GetSupportedAddons(specID string) (addons []string, err error)
@@ -98,12 +101,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/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

@@ -50,6 +50,7 @@ const (
 	LimitParallelWebsocketConnectionsPerIpFlag   = "limit-parallel-websocket-connections-per-ip"
 	LimitWebsocketIdleTimeFlag                   = "limit-websocket-connection-idle-time"
 	RateLimitRequestPerSecondFlag                = "rate-limit-requests-per-second"
+	SkipPolicyVerificationFlag                   = "skip-policy-verification"
 	// specification default flags
 	AllowMissingApisByDefaultFlagName = "allow-missing-apis-by-default"
 )

protocol/integration/mocks.go

@@ -39,7 +39,7 @@ func (m *mockConsumerStateTracker) RegisterForSpecUpdates(ctx context.Context, s
 	return nil
 }
 
-func (m *mockConsumerStateTracker) RegisterFinalizationConsensusForUpdates(context.Context, *finalizationconsensus.FinalizationConsensus) {
+func (m *mockConsumerStateTracker) RegisterFinalizationConsensusForUpdates(context.Context, *finalizationconsensus.FinalizationConsensus, bool) {
 }
 
 func (m *mockConsumerStateTracker) RegisterForDowntimeParamsUpdates(ctx context.Context, downtimeParamsUpdatable updaters.DowntimeParamsUpdatable) error {

protocol/lavasession/consumer_session_manager_test.go

@@ -232,7 +232,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/rpcconsumer/rpcconsumer.go

@@ -96,7 +96,7 @@ type ConsumerStateTrackerInf interface {
 	RegisterForVersionUpdates(ctx context.Context, version *protocoltypes.Version, versionValidator updaters.VersionValidationInf)
 	RegisterConsumerSessionManagerForPairingUpdates(ctx context.Context, consumerSessionManager *lavasession.ConsumerSessionManager, staticProvidersList []*lavasession.RPCProviderEndpoint)
 	RegisterForSpecUpdates(ctx context.Context, specUpdatable updaters.SpecUpdatable, endpoint lavasession.RPCEndpoint) error
-	RegisterFinalizationConsensusForUpdates(context.Context, *finalizationconsensus.FinalizationConsensus)
+	RegisterFinalizationConsensusForUpdates(context.Context, *finalizationconsensus.FinalizationConsensus, bool)
 	RegisterForDowntimeParamsUpdates(ctx context.Context, downtimeParamsUpdatable updaters.DowntimeParamsUpdatable) error
 	TxConflictDetection(ctx context.Context, finalizationConflict *conflicttypes.FinalizationConflict, responseConflict *conflicttypes.ResponseConflict, conflictHandler common.ConflictHandlerInterface) error
 	GetConsumerPolicy(ctx context.Context, consumerAddress, chainID string) (*plantypes.Policy, error)
@@ -348,6 +348,15 @@ func (rpcc *RPCConsumer) CreateConsumerEndpoint(
 		return nil, err
 	}
 
+	// Filter the relevant static providers
+	relevantStaticProviderList := []*lavasession.RPCProviderEndpoint{}
+	for _, staticProvider := range options.staticProvidersList {
+		if staticProvider.ChainID == rpcEndpoint.ChainID {
+			relevantStaticProviderList = append(relevantStaticProviderList, staticProvider)
+		}
+	}
+	staticProvidersActive := len(relevantStaticProviderList) > 0
+
 	_, averageBlockTime, _, _ := chainParser.ChainBlockStats()
 	var optimizer *provideroptimizer.ProviderOptimizer
 	var consumerConsistency *ConsumerConsistency
@@ -387,7 +396,7 @@ func (rpcc *RPCConsumer) CreateConsumerEndpoint(
 			return utils.LavaFormatError("failed loading finalization consensus", err, utils.LogAttr("endpoint", rpcEndpoint.Key()))
 		}
 		if !loaded { // when creating new finalization consensus instance we need to register it to updates
-			consumerStateTracker.RegisterFinalizationConsensusForUpdates(ctx, finalizationConsensus)
+			consumerStateTracker.RegisterFinalizationConsensusForUpdates(ctx, finalizationConsensus, staticProvidersActive)
 		}
 		return nil
 	}
@@ -771,6 +780,8 @@ 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")
+
 	cmdRPCConsumer.Flags().Bool(LavaOverLavaBackupFlagName, true, "enable lava over lava backup to regular rpc calls")
 	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")
 	common.AddRollingLogConfig(cmdRPCConsumer)

protocol/statetracker/consumer_state_tracker.go

@@ -94,8 +94,8 @@ func (cst *ConsumerStateTracker) RegisterForPairingUpdates(ctx context.Context,
 	}
 }
 
-func (cst *ConsumerStateTracker) RegisterFinalizationConsensusForUpdates(ctx context.Context, finalizationConsensus *finalizationconsensus.FinalizationConsensus) {
-	finalizationConsensusUpdater := updaters.NewFinalizationConsensusUpdater(cst.StateQuery, finalizationConsensus.SpecId)
+func (cst *ConsumerStateTracker) RegisterFinalizationConsensusForUpdates(ctx context.Context, finalizationConsensus *finalizationconsensus.FinalizationConsensus, ignoreQueryErrors bool) {
+	finalizationConsensusUpdater := updaters.NewFinalizationConsensusUpdater(cst.StateQuery, finalizationConsensus.SpecId, ignoreQueryErrors)
 	finalizationConsensusUpdaterRaw := cst.StateTracker.RegisterForUpdates(ctx, finalizationConsensusUpdater)
 	finalizationConsensusUpdater, ok := finalizationConsensusUpdaterRaw.(*updaters.FinalizationConsensusUpdater)
 	if !ok {

protocol/statetracker/updaters/finalization_consensus_updater.go

@@ -19,10 +19,11 @@ type FinalizationConsensusUpdater struct {
 	nextBlockForUpdate                uint64
 	stateQuery                        *ConsumerStateQuery
 	specId                            string
+	ignoreQueryErrors                 bool // used when static providers are configured so we don't spam errors on failed get pairing.
 }
 
-func NewFinalizationConsensusUpdater(stateQuery *ConsumerStateQuery, specId string) *FinalizationConsensusUpdater {
-	return &FinalizationConsensusUpdater{registeredFinalizationConsensuses: []*finalizationconsensus.FinalizationConsensus{}, stateQuery: stateQuery, specId: specId}
+func NewFinalizationConsensusUpdater(stateQuery *ConsumerStateQuery, specId string, ignoreQueryErrors bool) *FinalizationConsensusUpdater {
+	return &FinalizationConsensusUpdater{registeredFinalizationConsensuses: []*finalizationconsensus.FinalizationConsensus{}, stateQuery: stateQuery, specId: specId, ignoreQueryErrors: ignoreQueryErrors}
 }
 
 func (fcu *FinalizationConsensusUpdater) RegisterFinalizationConsensus(finalizationConsensus *finalizationconsensus.FinalizationConsensus) {
@@ -46,7 +47,9 @@ func (fcu *FinalizationConsensusUpdater) updateInner(latestBlock int64) {
 	defer cancel()
 	_, epoch, nextBlockForUpdate, err := fcu.stateQuery.GetPairing(timeoutCtx, fcu.specId, latestBlock)
 	if err != nil {
-		utils.LavaFormatError("could not get block stats for finalization consensus updater, trying again next block", err, utils.Attribute{Key: "latestBlock", Value: latestBlock})
+		if !fcu.ignoreQueryErrors {
+			utils.LavaFormatError("could not get block stats for finalization consensus updater, trying again next block", err, utils.Attribute{Key: "latestBlock", Value: latestBlock})
+		}
 		fcu.nextBlockForUpdate += 1
 		return
 	}

protocol/statetracker/updaters/pairing_updater.go

@@ -46,7 +46,7 @@ func NewPairingUpdater(stateQuery ConsumerStateQueryInf, specId string) *Pairing
 	return &PairingUpdater{consumerSessionManagersMap: map[string][]ConsumerSessionManagerInf{}, stateQuery: stateQuery, specId: specId, staticProviders: []*lavasession.RPCProviderEndpoint{}}
 }
 
-func (pu *PairingUpdater) updateStaticProviders(staticProviders []*lavasession.RPCProviderEndpoint) {
+func (pu *PairingUpdater) updateStaticProviders(staticProviders []*lavasession.RPCProviderEndpoint) int {
 	pu.lock.Lock()
 	defer pu.lock.Unlock()
 	if len(staticProviders) > 0 && len(pu.staticProviders) == 0 {
@@ -56,17 +56,30 @@ func (pu *PairingUpdater) updateStaticProviders(staticProviders []*lavasession.R
 			}
 		}
 	}
+	// return length of relevant static providers
+	return len(pu.staticProviders)
+}
+
+func (pu *PairingUpdater) getNumberOfStaticProviders() int {
+	pu.lock.RLock()
+	defer pu.lock.RUnlock()
+	return len(pu.staticProviders)
 }
 
 func (pu *PairingUpdater) RegisterPairing(ctx context.Context, consumerSessionManager ConsumerSessionManagerInf, staticProviders []*lavasession.RPCProviderEndpoint) error {
 	chainID := consumerSessionManager.RPCEndpoint().ChainID
 	timeoutCtx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 	pairingList, epoch, nextBlockForUpdate, err := pu.stateQuery.GetPairing(timeoutCtx, chainID, -1)
+	numberOfRelevantProviders := pu.updateStaticProviders(staticProviders)
+	// If user configured static providers but did not buy a subscription
 	if err != nil {
-		return err
+		if numberOfRelevantProviders == 0 {
+			return err
+		}
+		// else we continue with static providers.
+		epoch += 1
 	}
-	pu.updateStaticProviders(staticProviders)
 	pu.updateConsumerSessionManager(ctx, pairingList, consumerSessionManager, epoch)
 	if nextBlockForUpdate > pu.nextBlockForUpdate {
 		// make sure we don't update twice, this updates pu.nextBlockForUpdate
@@ -88,6 +101,10 @@ func (pu *PairingUpdater) RegisterPairingUpdatable(ctx context.Context, pairingU
 	defer pu.lock.Unlock()
 	_, epoch, _, err := pu.stateQuery.GetPairing(ctx, pu.specId, -1)
 	if err != nil {
+		if len(pu.staticProviders) > 0 {
+			// skipping errors for get pairing if static providers are set.
+			return nil
+		}
 		return err
 	}
 
@@ -114,6 +131,9 @@ func (pu *PairingUpdater) updateInner(latestBlock int64) {
 		pairingList, epoch, nextBlockForUpdate, err := pu.stateQuery.GetPairing(timeoutCtx, chainID, latestBlock)
 		cancel()
 		if err != nil {
+			if len(pu.staticProviders) > 0 {
+				return
+			}
 			utils.LavaFormatError("could not update pairing for chain, trying again next block", err, utils.Attribute{Key: "chain", Value: chainID})
 			nextBlockForUpdateList = append(nextBlockForUpdateList, pu.nextBlockForUpdate+1)
 			continue
@@ -192,10 +212,11 @@ func (pu *PairingUpdater) addStaticProvidersToPairingList(pairingList map[uint64
 			for _, extension := range url.Addons {
 				extensions[extension] = struct{}{}
 			}
+			// TODO might be problematic adding both addons and extensions with same map.
 			endpoint := &lavasession.Endpoint{
 				NetworkAddress: url.Url,
 				Enabled:        true,
-				Addons:         map[string]struct{}{}, // TODO: does not support addons, if required need to add the functionality to differentiate the two
+				Addons:         extensions,
 				Extensions:     extensions,
 				Connections:    []*lavasession.EndpointConnection{},
 			}
@@ -261,7 +282,7 @@ func (pu *PairingUpdater) filterPairingListByEndpoint(ctx context.Context, curre
 			totalStakeIncludingDelegation,
 		)
 	}
-	if len(pairing) == 0 {
+	if len(pairing)+pu.getNumberOfStaticProviders() == 0 {
 		return nil, utils.LavaFormatError("Failed getting pairing for consumer, pairing is empty", err, utils.Attribute{Key: "apiInterface", Value: rpcEndpoint.ApiInterface}, utils.Attribute{Key: "ChainID", Value: rpcEndpoint.ChainID}, utils.Attribute{Key: "geolocation", Value: rpcEndpoint.Geolocation})
 	}
 	// replace previous pairing with new providers

protocol/statetracker/updaters/state_query.go

@@ -183,7 +183,7 @@ func (csq *ConsumerStateQuery) GetPairing(ctx context.Context, chainID string, l
 		Client:  csq.fromAddress,
 	})
 	if err != nil {
-		return nil, 0, 0, utils.LavaFormatError("Failed in get pairing query", err, utils.Attribute{})
+		return nil, 0, 0, err
 	}
 	csq.lastChainID = chainID
 	csq.ResponsesCache.SetWithTTL(PairingRespKey+chainID, pairingResp, 1, DefaultTimeToLiveExpiration)

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},

scripts/pre_setups/init_lava_static_provider.sh → scripts/pre_setups/init_lava_static_provider_with_node.sh

@@ -16,8 +16,9 @@ make install-all
 echo "[Test Setup] setting up a new lava node"
 screen -d -m -S node bash -c "./scripts/start_env_dev.sh"
 screen -ls
-echo "[Lavavisor Setup] sleeping 20 seconds for node to finish setup (if its not enough increase timeout)"
-sleep 20
+echo "[Test Setup] sleeping 20 seconds for node to finish setup (if its not enough increase timeout)"
+sleep 5
+wait_for_lava_node_to_start
 
 GASPRICE="0.00002ulava"
 lavad tx gov submit-legacy-proposal spec-add ./cookbook/specs/ibc.json,./cookbook/specs/cosmoswasm.json,./cookbook/specs/tendermint.json,./cookbook/specs/cosmossdk.json,./cookbook/specs/cosmossdk_45.json,./cookbook/specs/cosmossdk_full.json,./cookbook/specs/ethermint.json,./cookbook/specs/ethereum.json,./cookbook/specs/cosmoshub.json,./cookbook/specs/lava.json,./cookbook/specs/osmosis.json,./cookbook/specs/fantom.json,./cookbook/specs/celo.json,./cookbook/specs/optimism.json,./cookbook/specs/arbitrum.json,./cookbook/specs/starknet.json,./cookbook/specs/aptos.json,./cookbook/specs/juno.json,./cookbook/specs/polygon.json,./cookbook/specs/evmos.json,./cookbook/specs/base.json,./cookbook/specs/canto.json,./cookbook/specs/sui.json,./cookbook/specs/solana.json,./cookbook/specs/bsc.json,./cookbook/specs/axelar.json,./cookbook/specs/avalanche.json,./cookbook/specs/fvm.json --lava-dev-test -y --from alice --gas-adjustment "1.5" --gas "auto" --gas-prices $GASPRICE &
@@ -47,9 +48,19 @@ lavad tx pairing stake-provider "LAV1" $PROVIDERSTAKE "$PROVIDER1_LISTENER,1"  1
 
 sleep_until_next_epoch
 
+screen -d -m -S provider1 bash -c "source ~/.bashrc; lavap rpcprovider \
+$PROVIDER1_LISTENER LAV1 rest '$LAVA_REST' \
+$PROVIDER1_LISTENER LAV1 tendermintrpc '$LAVA_RPC,$LAVA_RPC_WS' \
+$PROVIDER1_LISTENER LAV1 grpc '$LAVA_GRPC' \
+$EXTRA_PROVIDER_FLAGS --geolocation 1 --log_level trace --from servicer1 --chain-id lava --metrics-listen-address ":7776" 2>&1 | tee $LOGS_DIR/PROVIDER1.log" && sleep 0.25
+
+wait_next_block
+
 screen -d -m -S provider4 bash -c "source ~/.bashrc; lavap rpcprovider provider_examples/lava_example.yml\
 $EXTRA_PROVIDER_FLAGS --geolocation 1 --log_level debug --from servicer4 --static-providers --chain-id lava 2>&1 | tee $LOGS_DIR/PROVIDER4.log" && sleep 0.25
 
+wait_next_block
+
 screen -d -m -S consumers bash -c "source ~/.bashrc; lavap rpcconsumer consumer_examples/lava_consumer_static_peers.yml \
 $EXTRA_PORTAL_FLAGS --geolocation 1 --log_level debug --from user1 --chain-id lava --allow-insecure-provider-dialing --metrics-listen-address ":7779" 2>&1 | tee $LOGS_DIR/CONSUMERS.log" && sleep 0.25