[Relay] E2E Relay Gaps (#177)

Spike to close the E2E relay MVP.

---------

Co-authored-by: Redouane Lakrache <[email protected]>
Co-authored-by: Daniel Olshansky <[email protected]>
Co-authored-by: harry <[email protected]>

.gitignore

@@ -61,4 +61,12 @@ localnet_config.yaml
 # Relase artifacts produced by `ignite chain build --release`
 release
 
+# SMT KVStore files
+# TODO_TECHDEBT(#126, @red-0ne):  Rename `smt` to `smt_stores` and make it configurable so it can be stored anywhere on this
+smt
+
+# Do not allow a multi-moduled projected
 go.work.sum
+
+# TODO_TECHDEBT: It seems that .dot files come and go so we need to figure out the root cause: https://github.com/pokt-network/poktroll/pull/177/files#r1392521547
+# **/*.dot

Makefile

@@ -282,8 +282,9 @@ app_list: ## List all the staked applications
 app_stake: ## Stake tokens for the application specified (must specify the APP and SERVICES env vars)
 	poktrolld --home=$(POKTROLLD_HOME) tx application stake-application 1000upokt $(SERVICES) --keyring-backend test --from $(APP) --node $(POCKET_NODE)
 
+# TODO_IMPROVE(#180): Make sure genesis-staked actors are available via AccountKeeper
 .PHONY: app1_stake
-app1_stake: ## Stake app1
+app1_stake: ## Stake app1 (also staked in genesis)
 	APP=app1 SERVICES=anvil,svc1,svc2 make app_stake
 
 .PHONY: app2_stake
@@ -362,17 +363,24 @@ supplier_list: ## List all the staked supplier
 supplier_stake: ## Stake tokens for the supplier specified (must specify the APP env var)
 	poktrolld --home=$(POKTROLLD_HOME) tx supplier stake-supplier 1000upokt "$(SERVICES)" --keyring-backend test --from $(SUPPLIER) --node $(POCKET_NODE)
 
+# TODO_IMPROVE(#180): Make sure genesis-staked actors are available via AccountKeeper
 .PHONY: supplier1_stake
-supplier1_stake: ## Stake supplier1
-	SUPPLIER=supplier1 SERVICES="anvil;http://anvil:8547,svc1;http://localhost:8081" make supplier_stake
+supplier1_stake: ## Stake supplier1 (also staked in genesis)
+	# TODO_UPNEXT(@okdas): once `relayminer` service is added to tilt, this hostname should point to that service.
+	# I.e.: replace `localhost` with `relayminer` (or whatever the service's hostname is).
+	SUPPLIER=supplier1 SERVICES="anvil;http://localhost:8545,svc1;http://localhost:8081" make supplier_stake
 
 .PHONY: supplier2_stake
 supplier2_stake: ## Stake supplier2
-	SUPPLIER=supplier2 SERVICES="anvil;http://anvil:8547,svc2;http://localhost:8082" make supplier_stake
+	# TODO_UPNEXT(@okdas): once `relayminer` service is added to tilt, this hostname should point to that service.
+	# I.e.: replace `localhost` with `relayminer` (or whatever the service's hostname is).
+	SUPPLIER=supplier2 SERVICES="anvil;http://localhost:8545,svc2;http://localhost:8082" make supplier_stake
 
 .PHONY: supplier3_stake
 supplier3_stake: ## Stake supplier3
-	SUPPLIER=supplier3 SERVICES="anvil;http://anvil:8547,svc3;http://localhost:8083" make supplier_stake
+	# TODO_UPNEXT(@okdas): once `relayminer` service is added to tilt, this hostname should point to that service.
+	# I.e.: replace `localhost` with `relayminer` (or whatever the service's hostname is).
+	SUPPLIER=supplier3 SERVICES="anvil;http://localhost:8545,svc3;http://localhost:8083" make supplier_stake
 
 .PHONY: supplier_unstake
 supplier_unstake: ## Unstake an supplier (must specify the SUPPLIER env var)

cmd/pocketd/cmd/root.go

@@ -44,6 +44,7 @@ import (
 	"github.com/pokt-network/poktroll/app"
 	appparams "github.com/pokt-network/poktroll/app/params"
 	appgateservercmd "github.com/pokt-network/poktroll/pkg/appgateserver/cmd"
+	relayercmd "github.com/pokt-network/poktroll/pkg/relayer/cmd"
 )
 
 // NewRootCmd creates a new root command for a Cosmos SDK application
@@ -142,6 +143,11 @@ func initRootCmd(
 		addModuleInitFlags,
 	)
 
+	// add relayer command
+	rootCmd.AddCommand(
+		relayercmd.RelayerCmd(),
+	)
+
 	// add keybase, auxiliary RPC, query, and tx child commands
 	rootCmd.AddCommand(
 		rpc.StatusCommand(),

cmd/signals/on_exit.go

@@ -0,0 +1,23 @@
+package signals
+
+import (
+	"os"
+	"os/signal"
+)
+
+// GoOnExitSignal calls the given callback when the process receives an interrupt
+// or kill signal.
+func GoOnExitSignal(onInterrupt func()) {
+	go func() {
+		// Set up sigCh to receive when this process receives an interrupt or
+		// kill signal.
+		sigCh := make(chan os.Signal, 1)
+		signal.Notify(sigCh, os.Interrupt, os.Kill)
+
+		// Block until we receive an interrupt or kill signal (OS-agnostic)
+		<-sigCh
+
+		// Call the onInterrupt callback.
+		onInterrupt()
+	}()
+}

config.yml

@@ -100,6 +100,7 @@ genesis:
             - endpoints:
                 - configs: []
                   rpc_type: JSON_RPC
+                  # TODO_UPNEXT(@okdas): once `relayminer` service is added to tilt, this hostname should point to it instead of `localhost`.
                   url: http://localhost:8545
               service:
                 id: anvil

pkg/appgateserver/cmd/cmd.go

@@ -7,24 +7,25 @@ import (
 	"log"
 	"net/http"
 	"net/url"
-	"os"
-	"os/signal"
 
 	"cosmossdk.io/depinject"
 	cosmosclient "github.com/cosmos/cosmos-sdk/client"
-	"github.com/cosmos/cosmos-sdk/client/flags"
+	cosmosflags "github.com/cosmos/cosmos-sdk/client/flags"
 	"github.com/spf13/cobra"
 
+	"github.com/pokt-network/poktroll/cmd/signals"
 	"github.com/pokt-network/poktroll/pkg/appgateserver"
-	blockclient "github.com/pokt-network/poktroll/pkg/client/block"
-	eventsquery "github.com/pokt-network/poktroll/pkg/client/events_query"
+	"github.com/pokt-network/poktroll/pkg/deps/config"
 )
 
+// We're `explicitly omitting default` so that the appgateserver crashes if these aren't specified.
+const omittedDefaultFlagValue = "explicitly omitting default"
+
 var (
 	flagSigningKey        string
 	flagSelfSigning       bool
 	flagListeningEndpoint string
-	flagCometWebsocketUrl string
+	flagQueryNodeUrl      string
 )
 
 func AppGateServerCmd() *cobra.Command {
@@ -56,13 +57,15 @@ relays to the AppGate server and function as an Application, provided that:
 		RunE: runAppGateServer,
 	}
 
+	// Custom flags
 	cmd.Flags().StringVar(&flagSigningKey, "signing-key", "", "The name of the key that will be used to sign relays")
 	cmd.Flags().StringVar(&flagListeningEndpoint, "listening-endpoint", "http://localhost:42069", "The host and port that the appgate server will listen on")
-	cmd.Flags().StringVar(&flagCometWebsocketUrl, "comet-websocket-url", "ws://localhost:36657/websocket", "The URL of the comet websocket endpoint to communicate with the pocket blockchain")
 	cmd.Flags().BoolVar(&flagSelfSigning, "self-signing", false, "Whether the server should sign all incoming requests with its own ring (for applications)")
+	cmd.Flags().StringVar(&flagQueryNodeUrl, "query-node", omittedDefaultFlagValue, "tcp://<host>:<port> to a full pocket node for reading data and listening for on-chain events")
 
-	cmd.Flags().String(flags.FlagKeyringBackend, "", "Select keyring's backend (os|file|kwallet|pass|test)")
-	cmd.Flags().String(flags.FlagNode, "tcp://localhost:36657", "The URL of the comet tcp endpoint to communicate with the pocket blockchain")
+	// Cosmos flags
+	cmd.Flags().String(cosmosflags.FlagKeyringBackend, "", "Select keyring's backend (os|file|kwallet|pass|test)")
+	cmd.Flags().String(cosmosflags.FlagNode, omittedDefaultFlagValue, "registering the default cosmos node flag; needed to initialize the cosmostx and query contexts correctly and uses flagQueryNodeUrl underneath")
 
 	return cmd
 }
@@ -72,18 +75,8 @@ func runAppGateServer(cmd *cobra.Command, _ []string) error {
 	ctx, cancelCtx := context.WithCancel(cmd.Context())
 	defer cancelCtx()
 
-	// Handle interrupts in a goroutine.
-	go func() {
-		sigCh := make(chan os.Signal, 1)
-		signal.Notify(sigCh, os.Interrupt)
-
-		// Block until we receive an interrupt or kill signal (OS-agnostic)
-		<-sigCh
-		log.Println("INFO: Interrupt signal received, shutting down...")
-
-		// Signal goroutines to stop
-		cancelCtx()
-	}()
+	// Handle interrupt and kill signals asynchronously.
+	signals.GoOnExitSignal(cancelCtx)
 
 	// Parse the listening endpoint.
 	listeningUrl, err := url.Parse(flagListeningEndpoint)
@@ -92,7 +85,7 @@ func runAppGateServer(cmd *cobra.Command, _ []string) error {
 	}
 
 	// Setup the AppGate server dependencies.
-	appGateServerDeps, err := setupAppGateServerDependencies(cmd, ctx, flagCometWebsocketUrl)
+	appGateServerDeps, err := setupAppGateServerDependencies(ctx, cmd)
 	if err != nil {
 		return fmt.Errorf("failed to setup AppGate server dependencies: %w", err)
 	}
@@ -127,21 +120,62 @@ func runAppGateServer(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
-func setupAppGateServerDependencies(cmd *cobra.Command, ctx context.Context, cometWebsocketUrl string) (depinject.Config, error) {
-	// Retrieve the client context for the chain interactions.
-	clientCtx := cosmosclient.GetClientContextFromCmd(cmd)
+func setupAppGateServerDependencies(
+	ctx context.Context,
+	cmd *cobra.Command,
+) (depinject.Config, error) {
+	pocketNodeWebsocketUrl, err := getPocketNodeWebsocketUrl()
+	if err != nil {
+		return nil, err
+	}
 
-	// Create the events client.
-	eventsQueryClient := eventsquery.NewEventsQueryClient(flagCometWebsocketUrl)
+	supplierFuncs := []config.SupplierFn{
+		config.NewSupplyEventsQueryClientFn(pocketNodeWebsocketUrl),
+		config.NewSupplyBlockClientFn(pocketNodeWebsocketUrl),
+		newSupplyQueryClientContextFn(flagQueryNodeUrl),
+	}
 
-	// Create the block client.
-	log.Printf("INFO: Creating block client, using comet websocket URL: %s...", flagCometWebsocketUrl)
-	deps := depinject.Supply(eventsQueryClient)
-	blockClient, err := blockclient.NewBlockClient(ctx, deps, flagCometWebsocketUrl)
+	return config.SupplyConfig(ctx, cmd, supplierFuncs)
+}
+
+// getPocketNodeWebsocketUrl returns the websocket URL of the Pocket Node to
+// connect to for subscribing to on-chain events.
+func getPocketNodeWebsocketUrl() (string, error) {
+	if flagQueryNodeUrl == omittedDefaultFlagValue {
+		return "", errors.New("missing required flag: --query-node")
+	}
+
+	pocketNodeURL, err := url.Parse(flagQueryNodeUrl)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create block client: %w", err)
+		return "", err
 	}
 
-	// Return the dependencies config.
-	return depinject.Supply(clientCtx, blockClient), nil
+	return fmt.Sprintf("ws://%s/websocket", pocketNodeURL.Host), nil
+}
+
+// newSupplyQueryClientContextFn returns a new depinject.Config which is supplied with
+// the given deps and a new cosmos ClientCtx
+func newSupplyQueryClientContextFn(pocketQueryClientUrl string) config.SupplierFn {
+	return func(
+		_ context.Context,
+		deps depinject.Config,
+		cmd *cobra.Command,
+	) (depinject.Config, error) {
+		// Set --node flag to the pocketQueryClientUrl for the client context
+		// This flag is read by cosmosclient.GetClientQueryContext.
+		err := cmd.Flags().Set(cosmosflags.FlagNode, pocketQueryClientUrl)
+		if err != nil {
+			return nil, err
+		}
+
+		// Get the client context from the command.
+		queryClientCtx, err := cosmosclient.GetClientQueryContext(cmd)
+		if err != nil {
+			return nil, err
+		}
+		deps = depinject.Configs(deps, depinject.Supply(
+			queryClientCtx,
+		))
+		return deps, nil
+	}
 }

pkg/appgateserver/endpoint_selector.go

@@ -31,7 +31,7 @@ func (app *appGateServer) getRelayerUrl(
 				if endpoint.RpcType == rpcType {
 					supplierUrl, err := url.Parse(endpoint.Url)
 					if err != nil {
-						log.Printf("error parsing url: %s", err)
+						log.Printf("ERROR: error parsing url: %s", err)
 						continue
 					}
 					return supplierUrl, supplier.Address, nil

pkg/appgateserver/errors.go

@@ -10,4 +10,6 @@ var (
 	ErrAppGateMissingAppAddress             = sdkerrors.Register(codespace, 4, "missing application address")
 	ErrAppGateMissingSigningInformation     = sdkerrors.Register(codespace, 5, "missing app client signing information")
 	ErrAppGateMissingListeningEndpoint      = sdkerrors.Register(codespace, 6, "missing app client listening endpoint")
+	ErrAppGateEmptyRelayResponse            = sdkerrors.Register(codespace, 7, "empty relay response")
+	ErrAppGateHandleRelay                   = sdkerrors.Register(codespace, 8, "internal error handling relay request")
 )