Merge branch 'main' into PRT-near-shard-verification

README.md

@@ -30,7 +30,7 @@ The best way to start working with lava is to use docker, for additional reading
 
 ## Contributing
 
-See [CONTRIBUTING.md](./CONTRIBUTING.md) for details on how to contribute. If you want to follow the updates or learn more about the latest design then join our [Discord](https://discord.gg/lavanetxyz).
+See [CONTRIBUTING.md](./CONTRIBUTING.md) for details on how to contribute. If you want to follow the updates or learn more about the latest design then join our [Discord](https://discord.com/invite/Tbk5NxTCdA).
 
 ## Developing 
 
@@ -71,7 +71,7 @@ Or check out the latest [release](https://github.com/lavanet/lava/releases).
 
 ### Add `lavad`/`lavap` autocomplete
 
-You can add a useful autocomplete feature to `lavad` & `lavap` with a simple bash [script](https://github.com/lavanet/lava/blob/main/scripts/lava_auto_completion_install.sh).
+You can add a useful autocomplete feature to `lavad` & `lavap` with a simple bash [script](https://github.com/lavanet/lava/blob/main/scripts/automation_scripts/lava_auto_completion_install.sh).
 
 ## Join Lava
 
@@ -82,5 +82,5 @@ Join Lava's testnet, [read instructions here](https://docs.lavanet.xyz/testnet?u
 ## Community
 
 - [Github Discussions](https://github.com/lavanet/lava/discussions)
-- [Discord](https://discord.gg/lavanetxyz)
-- [Twitter](https://twitter.com/lavanetxyz)
+- [Discord](https://discord.com/invite/Tbk5NxTCdA)
+- [X (formerly Twitter)](https://x.com/lavanetxyz)

cookbook/specs/tendermint.json

@@ -139,7 +139,7 @@
                                 "compute_units": 10,
                                 "enabled": true,
                                 "category": {
-                                    "deterministic": true,
+                                    "deterministic": false,
                                     "local": false,
                                     "subscription": false,
                                     "stateful": 0
@@ -474,7 +474,7 @@
                                 "compute_units": 10,
                                 "enabled": true,
                                 "category": {
-                                    "deterministic": true,
+                                    "deterministic": false,
                                     "local": false,
                                     "subscription": false,
                                     "stateful": 0

protocol/chainlib/chain_message.go

@@ -31,13 +31,26 @@ type baseChainMessageContainer struct {
 	timeoutOverride        time.Duration
 	forceCacheRefresh      bool
 	parseDirective         *spectypes.ParseDirective // setting the parse directive related to the api, can be nil
+	usedDefaultValue       bool
 
 	inputHashCache []byte
 	// resultErrorParsingMethod passed by each api interface message to parse the result of the message
 	// and validate it doesn't contain a node error
 	resultErrorParsingMethod func(data []byte, httpStatusCode int) (hasError bool, errorMessage string)
 }
 
+func (bcmc *baseChainMessageContainer) UpdateEarliestInMessage(incomingEarliest int64) bool {
+	updatedSuccessfully := false
+	if bcmc.earliestRequestedBlock != spectypes.EARLIEST_BLOCK {
+		// check earliest is not unset (0) or incoming is lower than current value
+		if bcmc.earliestRequestedBlock == 0 || bcmc.earliestRequestedBlock > incomingEarliest {
+			bcmc.earliestRequestedBlock = incomingEarliest
+			updatedSuccessfully = true
+		}
+	}
+	return updatedSuccessfully
+}
+
 func (bcnc *baseChainMessageContainer) GetRequestedBlocksHashes() []string {
 	return bcnc.requestedBlockHashes
 }
@@ -158,6 +171,10 @@ func (bcnc *baseChainMessageContainer) OverrideExtensions(extensionNames []strin
 	}
 }
 
+func (bcnc *baseChainMessageContainer) GetUsedDefaultValue() bool {
+	return bcnc.usedDefaultValue
+}
+
 func (bcnc *baseChainMessageContainer) SetExtension(extension *spectypes.Extension) {
 	if len(bcnc.extensions) > 0 {
 		for _, ext := range bcnc.extensions {

protocol/chainlib/chainlib.go

@@ -91,6 +91,9 @@ type ChainMessage interface {
 	CheckResponseError(data []byte, httpStatusCode int) (hasError bool, errorMessage string)
 	GetRawRequestHash() ([]byte, error)
 	GetRequestedBlocksHashes() []string
+	UpdateEarliestInMessage(incomingEarliest int64) bool
+	SetExtension(extension *spectypes.Extension)
+	GetUsedDefaultValue() bool
 
 	ChainMessageForSend
 }

protocol/chainlib/extensionslib/extension_parser.go

@@ -71,7 +71,6 @@ func (ep *ExtensionParser) ExtensionParsing(addon string, extensionsChainMessage
 	if len(ep.configuredExtensions) == 0 {
 		return
 	}
-
 	for extensionKey, extension := range ep.configuredExtensions {
 		if extensionKey.Addon != addon {
 			// this extension is not relevant for this api

protocol/chainlib/grpc.go

@@ -115,7 +115,9 @@ func (apip *GrpcChainParser) CraftMessage(parsing *spectypes.ParseDirective, con
 	if err != nil {
 		return nil, err
 	}
-	return apip.newChainMessage(apiCont.api, spectypes.NOT_APPLICABLE, nil, grpcMessage, apiCollection), nil
+	parsedInput := &parser.ParsedInput{}
+	parsedInput.SetBlock(spectypes.NOT_APPLICABLE)
+	return apip.newChainMessage(apiCont.api, parsedInput, grpcMessage, apiCollection), nil
 }
 
 // ParseMsg parses message data into chain message object
@@ -165,18 +167,19 @@ func (apip *GrpcChainParser) ParseMsg(url string, data []byte, connectionType st
 				utils.LogAttr("overwriteRequestedBlock", overwriteReqBlock),
 			)
 			parsedInput.SetBlock(spectypes.NOT_APPLICABLE)
+		} else {
+			parsedInput.UsedDefaultValue = false
 		}
 	}
 
-	parsedBlock := parsedInput.GetBlock()
-	blockHashes, _ := parsedInput.GetBlockHashes()
-
-	nodeMsg := apip.newChainMessage(apiCont.api, parsedBlock, blockHashes, &grpcMessage, apiCollection)
+	nodeMsg := apip.newChainMessage(apiCont.api, parsedInput, &grpcMessage, apiCollection)
 	apip.BaseChainParser.ExtensionParsing(apiCollection.CollectionData.AddOn, nodeMsg, extensionInfo)
 	return nodeMsg, apip.BaseChainParser.Validate(nodeMsg)
 }
 
-func (*GrpcChainParser) newChainMessage(api *spectypes.Api, requestedBlock int64, requestedHashes []string, grpcMessage *rpcInterfaceMessages.GrpcMessage, apiCollection *spectypes.ApiCollection) *baseChainMessageContainer {
+func (*GrpcChainParser) newChainMessage(api *spectypes.Api, parsedInput *parser.ParsedInput, grpcMessage *rpcInterfaceMessages.GrpcMessage, apiCollection *spectypes.ApiCollection) *baseChainMessageContainer {
+	requestedBlock := parsedInput.GetBlock()
+	requestedHashes, _ := parsedInput.GetBlockHashes()
 	nodeMsg := &baseChainMessageContainer{
 		api:                      api,
 		msg:                      grpcMessage, // setting the grpc message as a pointer so we can set descriptors for parsing
@@ -185,6 +188,7 @@ func (*GrpcChainParser) newChainMessage(api *spectypes.Api, requestedBlock int64
 		apiCollection:            apiCollection,
 		resultErrorParsingMethod: grpcMessage.CheckResponseError,
 		parseDirective:           GetParseDirective(api, apiCollection),
+		usedDefaultValue:         parsedInput.UsedDefaultValue,
 	}
 	return nodeMsg
 }

protocol/chainlib/jsonRPC.go

@@ -88,7 +88,7 @@ func (apip *JsonRPCChainParser) CraftMessage(parsing *spectypes.ParseDirective,
 	if err != nil {
 		return nil, err
 	}
-	return apip.newChainMessage(apiCont.api, spectypes.NOT_APPLICABLE, nil, msg, apiCollection), nil
+	return apip.newChainMessage(apiCont.api, spectypes.NOT_APPLICABLE, nil, msg, apiCollection, false), nil
 }
 
 // this func parses message data into chain message object
@@ -111,6 +111,7 @@ func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType
 	var apiCollection *spectypes.ApiCollection
 	var latestRequestedBlock, earliestRequestedBlock int64 = 0, 0
 	blockHashes := []string{}
+	parsedDefault := true
 	for idx, msg := range msgs {
 		parsedInput := parser.NewParsedInput()
 		internalPath := ""
@@ -145,6 +146,9 @@ func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType
 			if hashes, err := parsedInput.GetBlockHashes(); err == nil {
 				blockHashes = append(blockHashes, hashes...)
 			}
+			if !parsedInput.UsedDefaultValue {
+				parsedDefault = false
+			}
 		} else {
 			parsedBlock, err := msg.ParseBlock(overwriteReqBlock)
 			parsedInput.SetBlock(parsedBlock)
@@ -154,6 +158,8 @@ func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType
 					utils.LogAttr("overwriteReqBlock", overwriteReqBlock),
 				)
 				parsedInput.SetBlock(spectypes.NOT_APPLICABLE)
+			} else {
+				parsedInput.UsedDefaultValue = false
 			}
 		}
 
@@ -205,9 +211,9 @@ func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType
 
 	var nodeMsg *baseChainMessageContainer
 	if len(msgs) == 1 {
-		nodeMsg = apip.newChainMessage(api, latestRequestedBlock, blockHashes, &msgs[0], apiCollection)
+		nodeMsg = apip.newChainMessage(api, latestRequestedBlock, blockHashes, &msgs[0], apiCollection, parsedDefault)
 	} else {
-		nodeMsg, err = apip.newBatchChainMessage(api, latestRequestedBlock, earliestRequestedBlock, blockHashes, msgs, apiCollection)
+		nodeMsg, err = apip.newBatchChainMessage(api, latestRequestedBlock, earliestRequestedBlock, blockHashes, msgs, apiCollection, parsedDefault)
 		if err != nil {
 			return nil, err
 		}
@@ -216,7 +222,7 @@ func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType
 	return nodeMsg, apip.BaseChainParser.Validate(nodeMsg)
 }
 
-func (*JsonRPCChainParser) newBatchChainMessage(serviceApi *spectypes.Api, requestedBlock int64, earliestRequestedBlock int64, requestedBlockHashes []string, msgs []rpcInterfaceMessages.JsonrpcMessage, apiCollection *spectypes.ApiCollection) (*baseChainMessageContainer, error) {
+func (*JsonRPCChainParser) newBatchChainMessage(serviceApi *spectypes.Api, requestedBlock int64, earliestRequestedBlock int64, requestedBlockHashes []string, msgs []rpcInterfaceMessages.JsonrpcMessage, apiCollection *spectypes.ApiCollection, usedDefaultValue bool) (*baseChainMessageContainer, error) {
 	batchMessage, err := rpcInterfaceMessages.NewBatchMessage(msgs)
 	if err != nil {
 		return nil, err
@@ -230,11 +236,12 @@ func (*JsonRPCChainParser) newBatchChainMessage(serviceApi *spectypes.Api, reque
 		earliestRequestedBlock:   earliestRequestedBlock,
 		resultErrorParsingMethod: rpcInterfaceMessages.CheckResponseErrorForJsonRpcBatch,
 		parseDirective:           nil,
+		usedDefaultValue:         usedDefaultValue,
 	}
 	return nodeMsg, err
 }
 
-func (*JsonRPCChainParser) newChainMessage(serviceApi *spectypes.Api, requestedBlock int64, requestedBlockHashes []string, msg *rpcInterfaceMessages.JsonrpcMessage, apiCollection *spectypes.ApiCollection) *baseChainMessageContainer {
+func (*JsonRPCChainParser) newChainMessage(serviceApi *spectypes.Api, requestedBlock int64, requestedBlockHashes []string, msg *rpcInterfaceMessages.JsonrpcMessage, apiCollection *spectypes.ApiCollection, usedDefaultValue bool) *baseChainMessageContainer {
 	nodeMsg := &baseChainMessageContainer{
 		api:                      serviceApi,
 		apiCollection:            apiCollection,
@@ -243,6 +250,7 @@ func (*JsonRPCChainParser) newChainMessage(serviceApi *spectypes.Api, requestedB
 		msg:                      msg,
 		resultErrorParsingMethod: msg.CheckResponseError,
 		parseDirective:           GetParseDirective(serviceApi, apiCollection),
+		usedDefaultValue:         usedDefaultValue,
 	}
 	return nodeMsg
 }

protocol/chainlib/protocol_message.go

@@ -3,8 +3,11 @@ package chainlib
 import (
 	"strings"
 
+	"github.com/lavanet/lava/v4/protocol/chainlib/extensionslib"
 	"github.com/lavanet/lava/v4/protocol/common"
+	"github.com/lavanet/lava/v4/utils"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
+	"github.com/lavanet/lava/v4/x/spec/types"
 )
 
 type UserData struct {
@@ -35,6 +38,48 @@ func (bpm *BaseProtocolMessage) HashCacheRequest(chainId string) ([]byte, func([
 	return HashCacheRequest(bpm.relayRequestData, chainId)
 }
 
+// addMissingExtensions adds any extensions from updatedProtocolExtensions that are not in currentPrivateDataExtensions
+func (bpm *BaseProtocolMessage) addMissingExtensions(updatedProtocolExtensions []*types.Extension, currentPrivateDataExtensions []string) []string {
+	// Create a map for O(1) lookups
+	existingExtensions := make(map[string]struct{}, len(currentPrivateDataExtensions))
+	for _, ext := range currentPrivateDataExtensions {
+		existingExtensions[ext] = struct{}{}
+	}
+
+	// Add missing extensions
+	for _, ext := range updatedProtocolExtensions {
+		if _, exists := existingExtensions[ext.Name]; !exists {
+			currentPrivateDataExtensions = append(currentPrivateDataExtensions, ext.Name)
+			if len(updatedProtocolExtensions) == len(currentPrivateDataExtensions) {
+				break
+			}
+		}
+	}
+	return currentPrivateDataExtensions
+}
+
+func (bpm *BaseProtocolMessage) UpdateEarliestAndValidateExtensionRules(extensionParser *extensionslib.ExtensionParser, earliestBlockHashRequested int64, addon string, seenBlock int64) bool {
+	if earliestBlockHashRequested >= 0 {
+		success := bpm.UpdateEarliestInMessage(earliestBlockHashRequested)
+		// check if we successfully updated the earliest block in the message
+		if success {
+			// parse the extensions for the new updated earliest block
+			extensionParser.ExtensionParsing(addon, bpm, uint64(seenBlock))
+			updatedProtocolExtensions := bpm.GetExtensions()
+			currentPrivateDataExtensions := bpm.RelayPrivateData().Extensions
+			utils.LavaFormatTrace("[Archive Debug] Trying to add extensions", utils.LogAttr("currentProtocolExtensions", updatedProtocolExtensions), utils.LogAttr("currentPrivateDataExtensions", currentPrivateDataExtensions))
+			if len(updatedProtocolExtensions) > len(currentPrivateDataExtensions) {
+				// we need to add the missing extension to the private data.
+				currentPrivateDataExtensions = bpm.addMissingExtensions(updatedProtocolExtensions, currentPrivateDataExtensions)
+				bpm.RelayPrivateData().Extensions = currentPrivateDataExtensions
+				utils.LavaFormatTrace("[Archive Debug] After Swap", utils.LogAttr("bpm.RelayPrivateData().Extensions", bpm.RelayPrivateData().Extensions))
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func (bpm *BaseProtocolMessage) GetBlockedProviders() []string {
 	if bpm.directiveHeaders == nil {
 		return nil
@@ -65,4 +110,5 @@ type ProtocolMessage interface {
 	HashCacheRequest(chainId string) ([]byte, func([]byte) []byte, error)
 	GetBlockedProviders() []string
 	GetUserData() common.UserData
+	UpdateEarliestAndValidateExtensionRules(extensionParser *extensionslib.ExtensionParser, earliestBlockHashRequested int64, addon string, seenBlock int64) bool
 }

protocol/chainlib/rest.go

@@ -69,12 +69,13 @@ func (apip *RestChainParser) CraftMessage(parsing *spectypes.ParseDirective, con
 	if err != nil {
 		return nil, err
 	}
-	api := apiCont.api
 	apiCollection, err := apip.getApiCollection(connectionType, apiCont.collectionKey.InternalPath, apiCont.collectionKey.Addon)
 	if err != nil {
 		return nil, err
 	}
-	return apip.newChainMessage(api, spectypes.NOT_APPLICABLE, nil, restMessage, apiCollection), nil
+	parsedInput := parser.NewParsedInput()
+	parsedInput.SetBlock(spectypes.NOT_APPLICABLE)
+	return apip.newChainMessage(apiCont.api, parsedInput, restMessage, apiCollection), nil
 }
 
 // ParseMsg parses message data into chain message object
@@ -126,26 +127,28 @@ func (apip *RestChainParser) ParseMsg(urlPath string, data []byte, connectionTyp
 				utils.LogAttr("overwriteRequestedBlock", overwriteReqBlock),
 			)
 			parsedInput.SetBlock(spectypes.NOT_APPLICABLE)
+		} else {
+			parsedInput.UsedDefaultValue = false
 		}
 	}
 
-	parsedBlock := parsedInput.GetBlock()
-	blockHashes, _ := parsedInput.GetBlockHashes()
-
-	nodeMsg := apip.newChainMessage(apiCont.api, parsedBlock, blockHashes, &restMessage, apiCollection)
+	nodeMsg := apip.newChainMessage(apiCont.api, parsedInput, &restMessage, apiCollection)
 	apip.BaseChainParser.ExtensionParsing(apiCollection.CollectionData.AddOn, nodeMsg, extensionInfo)
 	return nodeMsg, apip.BaseChainParser.Validate(nodeMsg)
 }
 
-func (*RestChainParser) newChainMessage(serviceApi *spectypes.Api, requestBlock int64, requestedHashes []string, restMessage *rpcInterfaceMessages.RestMessage, apiCollection *spectypes.ApiCollection) *baseChainMessageContainer {
+func (*RestChainParser) newChainMessage(api *spectypes.Api, parsedInput *parser.ParsedInput, restMessage *rpcInterfaceMessages.RestMessage, apiCollection *spectypes.ApiCollection) *baseChainMessageContainer {
+	requestedBlock := parsedInput.GetBlock()
+	requestedHashes, _ := parsedInput.GetBlockHashes()
 	nodeMsg := &baseChainMessageContainer{
-		api:                      serviceApi,
-		apiCollection:            apiCollection,
+		api:                      api,
 		msg:                      restMessage,
-		latestRequestedBlock:     requestBlock,
+		latestRequestedBlock:     requestedBlock,
 		requestedBlockHashes:     requestedHashes,
+		apiCollection:            apiCollection,
 		resultErrorParsingMethod: restMessage.CheckResponseError,
-		parseDirective:           GetParseDirective(serviceApi, apiCollection),
+		parseDirective:           GetParseDirective(api, apiCollection),
+		usedDefaultValue:         parsedInput.UsedDefaultValue,
 	}
 	return nodeMsg
 }