ARC-1-new-consumer-design

.github/workflows/e2e.yml

@@ -82,4 +82,4 @@ jobs:
     ### Run relayer unitests
     ######################################################
     - name: Run Relayer unit Tests
-      run: go test ./relayer/lavasession/ ./relayer/chainTracker/ -v
+      run: go test ./relayer/lavasession/ ./protocol/chaintracker/ -v

cmd/lavad/main.go

@@ -3,20 +3,25 @@ package main
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	_ "net/http/pprof"
 
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	"github.com/cosmos/cosmos-sdk/client/tx"
 	svrcmd "github.com/cosmos/cosmos-sdk/server/cmd"
+	"github.com/cosmos/cosmos-sdk/version"
 	"github.com/ignite-hq/cli/ignite/pkg/cosmoscmd"
 	"github.com/lavanet/lava/app"
+	"github.com/lavanet/lava/protocol/rpcconsumer"
 	"github.com/lavanet/lava/relayer"
 	"github.com/lavanet/lava/relayer/chainproxy"
+	"github.com/lavanet/lava/relayer/lavasession"
 	"github.com/lavanet/lava/relayer/performance"
 	"github.com/lavanet/lava/relayer/sentry"
 	"github.com/lavanet/lava/utils"
@@ -178,6 +183,43 @@ func main() {
 		},
 	}
 
+	cmdRPCClient := &cobra.Command{
+		Use:     "rpcconsumer [listen-ip]:[listen-port],[spec-chain-id],[api-interface] repeat...",
+		Short:   "rpcconsumer sets up a server to perform api requests and sends them through the lava protocol to data providers",
+		Long:    `rpcconsumer sets up a server to perform api requests and sends them through the lava protocol to data providers`,
+		Example: `rpcclient 127.0.0.1:3333,COS3,tendermintrpc 127.0.0.1:3334,COS3,rest`,
+		Args:    cobra.ExactArgs(4),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			utils.LavaFormatInfo("Gateway Proxy process started", &map[string]string{"args": strings.Join(args, ",")})
+			clientCtx, err := client.GetClientTxContext(cmd)
+			if err != nil {
+				return err
+			}
+
+			// decide on cli input method
+			// write good documentation for usage
+			// parse requested inputs into RPCEndpoint list
+			// handle flags, pass necessary fields
+			ctx := context.Background()
+			networkChainId, err := cmd.Flags().GetString(flags.FlagChainID)
+			if err != nil {
+				return err
+			}
+			txFactory := tx.NewFactoryCLI(clientCtx, cmd.Flags()).WithChainID(networkChainId)
+			rpcConsumer := rpcconsumer.RPCConsumer{}
+			rpcEndpoints := []*lavasession.RPCEndpoint{}
+			requiredResponses := 1 // TODO: handle secure flag, for a majority between providers
+			utils.LavaFormatInfo("lavad Binary Version: "+version.Version, nil)
+			rand.Seed(time.Now().UnixNano())
+			vrf_sk, _, err := utils.GetOrCreateVRFKey(clientCtx)
+			if err != nil {
+				utils.LavaFormatFatal("failed getting or creating a VRF key", err, nil)
+			}
+			rpcConsumer.Start(ctx, txFactory, clientCtx, rpcEndpoints, requiredResponses, vrf_sk)
+			return nil
+		},
+	}
+
 	flags.AddTxFlagsToCmd(cmdServer)
 	cmdServer.MarkFlagRequired(flags.FlagFrom)
 	flags.AddTxFlagsToCmd(cmdPortalServer)
@@ -205,6 +247,17 @@ func main() {
 	rootCmd.AddCommand(cmdPortalServer)
 	rootCmd.AddCommand(cmdTestClient)
 
+	// RPCConsumer command flags
+	flags.AddTxFlagsToCmd(cmdRPCClient)
+	cmdRPCClient.MarkFlagRequired(flags.FlagFrom)
+	cmdRPCClient.Flags().String(flags.FlagChainID, app.Name, "network chain id")
+	cmdRPCClient.Flags().Uint64(sentry.GeolocationFlag, 0, "geolocation to run from")
+	cmdRPCClient.MarkFlagRequired(sentry.GeolocationFlag)
+	cmdRPCClient.Flags().Bool("secure", false, "secure sends reliability on every message")
+	cmdRPCClient.Flags().String(performance.PprofAddressFlagName, "", "pprof server address, used for code profiling")
+	cmdRPCClient.Flags().String(performance.CacheFlagName, "", "address for a cache server to improve performance")
+	// rootCmd.AddCommand(cmdRPCClient) // Remove this when ready
+
 	if err := svrcmd.Execute(rootCmd, app.DefaultNodeHome); err != nil {
 		os.Exit(1)
 	}

protocol/chainlib/chainlib.go

@@ -0,0 +1,68 @@
+package chainlib
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lavanet/lava/relayer/lavasession"
+	pairingtypes "github.com/lavanet/lava/x/pairing/types"
+	spectypes "github.com/lavanet/lava/x/spec/types"
+)
+
+func NewChainParser(apiInterface string) (chainParser ChainParser, err error) {
+	switch apiInterface {
+	case spectypes.APIInterfaceJsonRPC:
+		return NewJrpcChainParser()
+	case spectypes.APIInterfaceTendermintRPC:
+		return NewTendermintRpcChainParser()
+	case spectypes.APIInterfaceRest:
+		return NewRestChainParser()
+	case spectypes.APIInterfaceGrpc:
+		return NewGrpcChainParser()
+	}
+	return nil, fmt.Errorf("chainParser for apiInterface (%s) not found", apiInterface)
+}
+
+func NewChainListener(ctx context.Context, listenEndpoint *lavasession.RPCEndpoint, relaySender RelaySender) (ChainListener, error) {
+	switch listenEndpoint.ApiInterface {
+	case spectypes.APIInterfaceJsonRPC:
+		return NewJrpcChainListener(ctx, listenEndpoint, relaySender), nil
+	case spectypes.APIInterfaceTendermintRPC:
+		return NewTendermintRpcChainListener(ctx, listenEndpoint, relaySender), nil
+	case spectypes.APIInterfaceRest:
+		return NewRestChainListener(ctx, listenEndpoint, relaySender), nil
+	case spectypes.APIInterfaceGrpc:
+		return NewGrpcChainListener(ctx, listenEndpoint, relaySender), nil
+	}
+	return nil, fmt.Errorf("chainListener for apiInterface (%s) not found", listenEndpoint.ApiInterface)
+}
+
+// this is an interface for parsing and generating messages of the supported APIType
+// it checks for the existence of the method in the spec, and formats the message
+type ChainParser interface {
+	ParseMsg(url string, data []byte, connectionType string) (ChainMessage, error) // has to be thread safe
+	SetSpec(spec spectypes.Spec)                                                   // has to be thread safe
+	DataReliabilityParams() (enabled bool, dataReliabilityThreshold uint32)
+	ChainBlockStats() (allowedBlockLagForQosSync int64, averageBlockTime time.Duration, blockDistanceForFinalizedData uint32)
+}
+
+type ChainMessage interface {
+	GetServiceApi() *spectypes.ServiceApi
+	GetInterface() *spectypes.ApiInterface
+	RequestedBlock() int64
+}
+
+type RelaySender interface {
+	SendRelay(
+		ctx context.Context,
+		url string,
+		req string,
+		connectionType string,
+		dappID string,
+	) (*pairingtypes.RelayReply, *pairingtypes.Relayer_RelaySubscribeClient, error)
+}
+
+type ChainListener interface {
+	Serve()
+}

protocol/chainlib/grpc.go

@@ -0,0 +1,41 @@
+package chainlib
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lavanet/lava/relayer/lavasession"
+	spectypes "github.com/lavanet/lava/x/spec/types"
+)
+
+type GrpcChainParser struct{}
+
+func (apip *GrpcChainParser) ParseMsg(url string, data []byte, connectionType string) (ChainMessage, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (apip *GrpcChainParser) SetSpec(spec spectypes.Spec) {}
+
+func (apip *GrpcChainParser) DataReliabilityParams() (enabled bool, dataReliabilityThreshold uint32) {
+	// TODO
+	return false, 0
+}
+
+func (apip *GrpcChainParser) ChainBlockStats() (allowedBlockLagForQosSync int64, averageBlockTime time.Duration, blockDistanceForFinalizedData uint32) {
+	return 0, 0, 0
+}
+
+func NewGrpcChainParser() (chainParser *GrpcChainParser, err error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+type GrpcChainListener struct{}
+
+func (apil *GrpcChainListener) Serve() {}
+
+func NewGrpcChainListener(ctx context.Context, listenEndpoint *lavasession.RPCEndpoint, relaySender RelaySender) (chainListener *GrpcChainListener) {
+	// open up server for grpc implementing the api requested (currently implemented in serve_portal in chainproxy, endpoint at listenEndpoint
+	// when receiving the data such as url, rpc data, headers (connectionType), use relaySender to wrap verify and send that data
+	return nil
+}

protocol/chainlib/jsonRPC.go

@@ -0,0 +1,42 @@
+package chainlib
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lavanet/lava/relayer/lavasession"
+	spectypes "github.com/lavanet/lava/x/spec/types"
+)
+
+type JsonRPCChainParser struct{}
+
+func (apip *JsonRPCChainParser) ParseMsg(url string, data []byte, connectionType string) (ChainMessage, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (apip *JsonRPCChainParser) SetSpec(spec spectypes.Spec) {}
+
+func (apip *JsonRPCChainParser) DataReliabilityParams() (enabled bool, dataReliabilityThreshold uint32) {
+	// TODO
+	return false, 0
+}
+
+func (apip *JsonRPCChainParser) ChainBlockStats() (allowedBlockLagForQosSync int64, averageBlockTime time.Duration, blockDistanceForFinalizedData uint32) {
+	// TODO:
+	return 0, 0, 0
+}
+
+func NewJrpcChainParser() (chainParser *JsonRPCChainParser, err error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+type JsonRPCChainListener struct{}
+
+func (apil *JsonRPCChainListener) Serve() {}
+
+func NewJrpcChainListener(ctx context.Context, listenEndpoint *lavasession.RPCEndpoint, relaySender RelaySender) (chainListener *JsonRPCChainListener) {
+	// open up server for http implementing the api requested (currently implemented in serve_portal in chainproxy, endpoint at listenEndpoint
+	// when receiving the data such as url, rpc data, headers (connectionType), use relaySender to wrap verify and send that data
+	return nil
+}

protocol/chainlib/rest.go

@@ -0,0 +1,42 @@
+package chainlib
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lavanet/lava/relayer/lavasession"
+	spectypes "github.com/lavanet/lava/x/spec/types"
+)
+
+type RestChainParser struct{}
+
+func (apip *RestChainParser) ParseMsg(url string, data []byte, connectionType string) (ChainMessage, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (apip *RestChainParser) SetSpec(spec spectypes.Spec) {}
+
+func (apip *RestChainParser) DataReliabilityParams() (enabled bool, dataReliabilityThreshold uint32) {
+	// TODO
+	return false, 0
+}
+
+func (apip *RestChainParser) ChainBlockStats() (allowedBlockLagForQosSync int64, averageBlockTime time.Duration, blockDistanceForFinalizedData uint32) {
+	// TODO:
+	return 0, 0, 0
+}
+
+func NewRestChainParser() (chainParser *RestChainParser, err error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+type RestChainListener struct{}
+
+func (apil *RestChainListener) Serve() {}
+
+func NewRestChainListener(ctx context.Context, listenEndpoint *lavasession.RPCEndpoint, relaySender RelaySender) (chainListener *RestChainListener) {
+	// open up server for http implementing the api requested (currently implemented in serve_portal in chainproxy, endpoint at listenEndpoint
+	// when receiving the data such as url, rpc data, headers (connectionType), use relaySender to wrap verify and send that data
+	return nil
+}

protocol/chainlib/tendermintRPC.go

@@ -0,0 +1,42 @@
+package chainlib
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/lavanet/lava/relayer/lavasession"
+	spectypes "github.com/lavanet/lava/x/spec/types"
+)
+
+type TendermintChainParser struct{}
+
+func (apip *TendermintChainParser) ParseMsg(url string, data []byte, connectionType string) (ChainMessage, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+func (apip *TendermintChainParser) SetSpec(spec spectypes.Spec) {}
+
+func (apip *TendermintChainParser) DataReliabilityParams() (enabled bool, dataReliabilityThreshold uint32) {
+	// TODO
+	return false, 0
+}
+
+func (apip *TendermintChainParser) ChainBlockStats() (allowedBlockLagForQosSync int64, averageBlockTime time.Duration, blockDistanceForFinalizedData uint32) {
+	// TODO:
+	return 0, 0, 0
+}
+
+func NewTendermintRpcChainParser() (chainParser *TendermintChainParser, err error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
+type TendermintRpcChainListener struct{}
+
+func (apil *TendermintRpcChainListener) Serve() {}
+
+func NewTendermintRpcChainListener(ctx context.Context, listenEndpoint *lavasession.RPCEndpoint, relaySender RelaySender) (chainListener *TendermintRpcChainListener) {
+	// open up server for http implementing the api requested (currently implemented in serve_portal in chainproxy, endpoint at listenEndpoint
+	// when receiving the data such as url, rpc data, headers (connectionType), use relaySender to wrap verify and send that data
+	return nil
+}