[Proxy] feat: implement relayer proxy full workflow

pkg/relayer/proxy/error_reply.go

@@ -8,9 +8,9 @@ import (
 )
 
 // replyWithError builds a JSONRPCResponseError from the passed in error and writes it to the writer.
-// TODO_IMPROVE: This method should be aware of the request id and use it in the response by having
+// TODO_TECHDEBT: This method should be aware of the request id and use it in the response by having
 // the caller pass it along with the error if available.
-// TODO_IMPROVE: This method should be aware of the nature of the error to use the appropriate JSONRPC
+// TODO_TECHDEBT: This method should be aware of the nature of the error to use the appropriate JSONRPC
 // Code, Message and Data. Possibly by augmenting the passed in error with the adequate information.
 func (j *jsonRPCServer) replyWithError(writer http.ResponseWriter, err error) {
 	relayResponse := &types.RelayResponse{
@@ -19,6 +19,7 @@ func (j *jsonRPCServer) replyWithError(writer http.ResponseWriter, err error) {
 				Id:      make([]byte, 0),
 				Jsonrpc: "2.0",
 				Error: &types.JSONRPCResponseError{
+					// Using conventional error code indicating internal server error.
 					Code:    -32000,
 					Message: err.Error(),
 					Data:    nil,
@@ -33,8 +34,7 @@ func (j *jsonRPCServer) replyWithError(writer http.ResponseWriter, err error) {
 		return
 	}
 
-	_, err = writer.Write(relayResponseBz)
-	if err != nil {
+	if _, err = writer.Write(relayResponseBz); err != nil {
 		log.Printf("ERROR: failed writing relay response: %s", err)
 		return
 	}

pkg/relayer/proxy/errors.go

@@ -1,11 +1,11 @@
 package proxy
 
-import errorsmod "cosmossdk.io/errors"
+import sdkerrors "cosmossdk.io/errors"
 
 var (
-	ErrUnsupportedRPCType = errorsmod.Register(codespace, 1, "unsupported rpc type")
-	ErrInvalidSignature   = errorsmod.Register(codespace, 2, "invalid signature")
-	ErrInvalidSession     = errorsmod.Register(codespace, 3, "invalid session")
-	ErrInvalidSupplier    = errorsmod.Register(codespace, 4, "invalid supplier")
-	codespace             = "relayer/proxy"
+	codespace                       = "relayer/proxy"
+	ErrUnsupportedRPCType           = sdkerrors.Register(codespace, 1, "unsupported rpc type")
+	ErrInvalidRelayRequestSignature = sdkerrors.Register(codespace, 2, "invalid relay request signature")
+	ErrInvalidSession               = sdkerrors.Register(codespace, 3, "invalid session")
+	ErrInvalidSupplier              = sdkerrors.Register(codespace, 4, "invalid supplier")
 )

pkg/relayer/proxy/interface.go

@@ -5,6 +5,7 @@ import (
 
 	"pocket/pkg/observable"
 	"pocket/x/service/types"
+	sharedtypes "pocket/x/shared/types"
 )
 
 // RelayerProxy is the interface for the proxy that serves relays to the application.
@@ -26,7 +27,7 @@ type RelayerProxy interface {
 
 	// VerifyRelayRequest is a shared method used by RelayServers to check the
 	// relay request signature and session validity.
-	VerifyRelayRequest(ctx context.Context, relayRequest *types.RelayRequest, serviceId string) error
+	VerifyRelayRequest(ctx context.Context, relayRequest *types.RelayRequest, serviceId *sharedtypes.ServiceId) error
 
 	// SignRelayResponse is a shared method used by RelayServers to sign the relay response.
 	SignRelayResponse(relayResponse *types.RelayResponse) ([]byte, error)
@@ -41,5 +42,5 @@ type RelayServer interface {
 	Stop(ctx context.Context) error
 
 	// ServiceId returns the serviceId of the service.
-	ServiceId() string
+	ServiceId() *sharedtypes.ServiceId
 }

pkg/relayer/proxy/jsonrpc.go

@@ -4,23 +4,26 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"log"
 	"net/http"
 	"net/url"
 
 	"pocket/x/service/types"
+	sharedtypes "pocket/x/shared/types"
 )
 
 var _ RelayServer = (*jsonRPCServer)(nil)
 
 type jsonRPCServer struct {
-	// serviceId is the identifier of the service that the server is responsible for.
-	serviceId string
+	// serviceId is the id of the service that the server is responsible for.
+	serviceId *sharedtypes.ServiceId
 
-	// endpointUrl is the URL that the server listens to for incoming relay requests.
-	endpointUrl string
+	// serverEndpoint is the advertised endpoint configuration that the server uses to
+	// listen for incoming relay requests.
+	serverEndpoint *sharedtypes.SupplierEndpoint
 
-	// nativeServiceListenAddress is the address of the native service to which the server relays requests.
-	nativeServiceListenAddress string
+	// proxiedServiceEndpoint is the address of the proxied service that the server relays requests to.
+	proxiedServiceEndpoint url.URL
 
 	// server is the HTTP server that listens for incoming relay requests.
 	server *http.Server
@@ -34,87 +37,92 @@ type jsonRPCServer struct {
 }
 
 // NewJSONRPCServer creates a new HTTP server that listens for incoming relay requests
-// and relays them to the supported native service.
+// and forwards them to the supported proxied service endpoint.
 // It takes the serviceId, endpointUrl, and the main RelayerProxy as arguments and returns
 // a RelayServer that listens to incoming RelayRequests.
 func NewJSONRPCServer(
-	serviceId string,
-	endpointUrl string,
-	nativeServiceListenAddress string,
+	serviceId *sharedtypes.ServiceId,
+	supplierEndpoint *sharedtypes.SupplierEndpoint,
+	proxiedServiceEndpoint url.URL,
 	servedRelaysProducer chan<- *types.Relay,
 	proxy RelayerProxy,
 ) RelayServer {
 	return &jsonRPCServer{
-		serviceId:                  serviceId,
-		endpointUrl:                endpointUrl,
-		server:                     &http.Server{Addr: endpointUrl},
-		relayerProxy:               proxy,
-		nativeServiceListenAddress: nativeServiceListenAddress,
-		servedRelaysProducer:       servedRelaysProducer,
+		serviceId:              serviceId,
+		serverEndpoint:         supplierEndpoint,
+		server:                 &http.Server{Addr: supplierEndpoint.Url},
+		relayerProxy:           proxy,
+		proxiedServiceEndpoint: proxiedServiceEndpoint,
+		servedRelaysProducer:   servedRelaysProducer,
 	}
 }
 
 // Start starts the service server and returns an error if it fails.
-// It also waits for the passed-in context to be done in order to shut down.
+// It also waits for the passed in context to end before shutting down.
 // This method is blocking and should be called in a goroutine.
-func (j *jsonRPCServer) Start(ctx context.Context) error {
+func (jsrv *jsonRPCServer) Start(ctx context.Context) error {
 	go func() {
 		<-ctx.Done()
-		j.server.Shutdown(ctx)
+		jsrv.server.Shutdown(ctx)
 	}()
 
-	return j.server.ListenAndServe()
+	return jsrv.server.ListenAndServe()
 }
 
 // Stop terminates the service server and returns an error if it fails.
-func (j *jsonRPCServer) Stop(ctx context.Context) error {
-	return j.server.Shutdown(ctx)
+func (jsrv *jsonRPCServer) Stop(ctx context.Context) error {
+	return jsrv.server.Shutdown(ctx)
 }
 
-// ServiceId returns the serviceId of the service.
-func (j *jsonRPCServer) ServiceId() string {
-	return j.serviceId
+// ServiceId returns the serviceId of the JSON-RPC service.
+func (jsrv *jsonRPCServer) ServiceId() *sharedtypes.ServiceId {
+	return jsrv.serviceId
 }
 
 // ServeHTTP listens for incoming relay requests. It implements the respective
 // method of the http.Handler interface. It is called by http.ListenAndServe()
 // when jsonRPCServer is used as an http.Handler with an http.Server.
 // (see https://pkg.go.dev/net/http#Handler)
-func (j *jsonRPCServer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+func (jsrv *jsonRPCServer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	ctx := request.Context()
-	// Relay the request to the native service and build the response that will be sent back to the client.
-	relay, err := j.serveHTTP(ctx, request)
+	// Relay the request to the proxied service and build the response that will be sent back to the client.
+	relay, err := jsrv.serveHTTP(ctx, request)
 	if err != nil {
 		// Reply with an error if relay response could not be built.
-		j.replyWithError(writer, err)
+		jsrv.replyWithError(writer, err)
+		log.Printf("WARN: failed serving relay request: %s", err)
 		return
 	}
 
 	// Send the relay response to the client.
-	if err := j.sendRelayResponse(relay.Res, writer); err != nil {
-		j.replyWithError(writer, err)
+	if err := jsrv.sendRelayResponse(relay.Res, writer); err != nil {
+		jsrv.replyWithError(writer, err)
+		log.Print("WARN: failed sending relay response: %s", err)
 		return
 	}
 
+	log.Print("INFO: relay request served successfully")
+
 	// Emit the relay to the servedRelays observable.
-	j.servedRelaysProducer <- relay
+	jsrv.servedRelaysProducer <- relay
 }
 
 // serveHTTP holds the underlying logic of ServeHTTP.
-func (j *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*types.Relay, error) {
+func (jsrv *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*types.Relay, error) {
 	// Extract the relay request from the request body.
-	relayRequest, err := j.newRelayRequest(request)
+	relayRequest, err := jsrv.newRelayRequest(request)
 	if err != nil {
 		return nil, err
 	}
 
 	// Verify the relay request signature and session.
-	if err := j.relayerProxy.VerifyRelayRequest(ctx, relayRequest, j.serviceId); err != nil {
+	if err := jsrv.relayerProxy.VerifyRelayRequest(ctx, relayRequest, jsrv.serviceId); err != nil {
 		return nil, err
 	}
 
-	// Get the relayRequest payload's ReadCloser to add it to the http.Request
+	// Get the relayRequest payload's `io.ReadCloser` to add it to the http.Request
 	// that will be sent to the native service.
+	// (see https://pkg.go.dev/net/http#Request) Body field type.
 	var payloadBz []byte
 	if _, err = relayRequest.Payload.MarshalTo(payloadBz); err != nil {
 		return nil, err
@@ -127,7 +135,7 @@ func (j *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*
 	if err != nil {
 		return nil, err
 	}
-	destinationURL.Host = j.nativeServiceListenAddress
+	destinationURL.Host = jsrv.proxiedServiceEndpoint.Host
 
 	relayHTTPRequest := &http.Request{
 		Method: request.Method,
@@ -146,7 +154,7 @@ func (j *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*
 	// Build the relay response from the native service response
 	// Use relayRequest.Meta.SessionHeader on the relayResponse session header since it was verified to be valid
 	// and has to be the same as the relayResponse session header.
-	relayResponse, err := j.newRelayResponse(httpResponse, relayRequest.Meta.SessionHeader)
+	relayResponse, err := jsrv.newRelayResponse(httpResponse, relayRequest.Meta.SessionHeader)
 	if err != nil {
 		return nil, err
 	}
@@ -156,15 +164,11 @@ func (j *jsonRPCServer) serveHTTP(ctx context.Context, request *http.Request) (*
 
 // sendRelayResponse marshals the relay response and sends it to the client.
 func (j *jsonRPCServer) sendRelayResponse(relayResponse *types.RelayResponse, writer http.ResponseWriter) error {
-	relayResposeBz, err := relayResponse.Marshal()
-	if err != nil {
-		return err
-	}
-
-	_, err = writer.Write(relayResposeBz)
+	relayResponseBz, err := relayResponse.Marshal()
 	if err != nil {
 		return err
 	}
 
-	return nil
+	_, err = writer.Write(relayResponseBz)
+	return err
 }

pkg/relayer/proxy/proxy.go

@@ -2,13 +2,14 @@
 
 import (
 	"context"
+	"net/url"
 
 	sdkclient "github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
 	accounttypes "github.com/cosmos/cosmos-sdk/x/auth/types"
 	"golang.org/x/sync/errgroup"
 
 	blocktypes "pocket/pkg/client"
 	"pocket/pkg/observable"
 	"pocket/pkg/observable/channel"
 	"pocket/x/service/types"
@@ -19,8 +20,9 @@
 var _ RelayerProxy = (*relayerProxy)(nil)
 
 type (
-	serviceId       = string
-	RelayServersMap = map[serviceId][]RelayServer
+	serviceId            = string
+	relayServersMap      = map[serviceId][]RelayServer
+	servicesEndpointsMap = map[serviceId]url.URL
 )
 
 type relayerProxy struct {
@@ -47,12 +49,11 @@
 
 	// advertisedRelayServers is a map of the services provided by the relayer proxy. Each provided service
 	// has the necessary information to start the server that listens for incoming relay requests and
-	// the client that relays the request to the supported native service.
-	advertisedRelayServers RelayServersMap
+	// the client that relays the request to the supported proxied service.
+	advertisedRelayServers relayServersMap
 
-	// nativeServiceListenAddresses is a map of the native services server's listen addresses
-	// that are supported by the relayer proxy.
-	nativeServicesListenAddress map[serviceId]string
+	// proxiedServicesEndpoints is a map of the proxied services endpoints that the relayer proxy supports.
+	proxiedServicesEndpoints servicesEndpointsMap
 
 	// servedRelays is an observable that notifies the miner about the relays that have been served.
 	servedRelays observable.Observable[*types.Relay]
@@ -73,7 +74,7 @@
 	clientCtx sdkclient.Context,
 	keyName string,
 	keyring keyring.Keyring,
-	nativeServicesListenAddress map[serviceId]string,
+	proxiedServicesEndpoints servicesEndpointsMap,
 	blockClient blocktypes.BlockClient,
 ) RelayerProxy {
 	accountQuerier := accounttypes.NewQueryClient(clientCtx)
@@ -82,50 +83,51 @@
 	servedRelays, servedRelaysProducer := channel.NewObservable[*types.Relay]()
 
 	return &relayerProxy{
-		blockClient:                 blockClient,
-		keyName:                     keyName,
-		keyring:                     keyring,
-		accountsQuerier:             accountQuerier,
-		supplierQuerier:             supplierQuerier,
-		sessionQuerier:              sessionQuerier,
-		nativeServicesListenAddress: nativeServicesListenAddress,
-		servedRelays:                servedRelays,
-		servedRelaysProducer:        servedRelaysProducer,
-		clientCtx:                   clientCtx,
+		blockClient:              blockClient,
+		keyName:                  keyName,
+		keyring:                  keyring,
+		accountsQuerier:          accountQuerier,
+		supplierQuerier:          supplierQuerier,
+		sessionQuerier:           sessionQuerier,
+		proxiedServicesEndpoints: proxiedServicesEndpoints,
+		servedRelays:             servedRelays,
+		servedRelaysProducer:     servedRelaysProducer,
+		clientCtx:                clientCtx,
 	}
 }
 
 // Start concurrently starts all advertised relay servers and returns an error if any of them fails to start.
+// This method is blocking until all RelayServers are started.
 func (rp *relayerProxy) Start(ctx context.Context) error {
 	// The provided services map is built from the supplier's on-chain advertised information,
 	// which is a runtime parameter that can be changed by the supplier.
-	// Build the provided services map at Start instead of NewRelayerProxy to avoid having to
+	// NOTE: We build the provided services map at Start instead of NewRelayerProxy to avoid having to
 	// return an error from the constructor.
-	err := rp.BuildProvidedServices(ctx)
-	if err != nil {
+	if err := rp.BuildProvidedServices(ctx); err != nil {
 		return err
 	}
 
-	startGroup, gctx := errgroup.WithContext(ctx)
+	startGroup, ctx := errgroup.WithContext(ctx)
 
 	for _, relayServer := range rp.advertisedRelayServers {
 		for _, svr := range relayServer {
 			server := svr // create a new variable scoped to the anonymous function
-			startGroup.Go(func() error { return server.Start(gctx) })
+			startGroup.Go(func() error { return server.Start(ctx) })
 		}
 	}
 
 	return startGroup.Wait()
 }
 
 // Stop concurrently stops all advertised relay servers and returns an error if any of them fails.
+// This method is blocking until all RelayServers are stopped.
 func (rp *relayerProxy) Stop(ctx context.Context) error {
-	stopGroup, gctx := errgroup.WithContext(ctx)
+	stopGroup, ctx := errgroup.WithContext(ctx)
 
 	for _, providedService := range rp.advertisedRelayServers {
 		for _, svr := range providedService {
 			server := svr // create a new variable scoped to the anonymous function
-			stopGroup.Go(func() error { return server.Stop(gctx) })
+			stopGroup.Go(func() error { return server.Stop(ctx) })
 		}
 	}