Merge pull request #327 from ava-labs/gstuart/api-relay

API endpoint for manually relaying warp message

README.md

@@ -316,6 +316,54 @@ The relayer consists of the following components:
   <img src="resources/relayer-diagram.png?raw=true"></img>
 </div>
 
+### API
+
+#### `/relay`
+- Used to manually relay a Warp message. The body of the request must contain the following JSON:
+```json
+{
+ "blockchain-id": "<cb58-encoding of blockchain ID>",
+ "message-id": "<cb58-encoding of Warp message ID>",
+ "block-num": "<Block number that the message was sent in>"
+}
+```
+- If successful, the endpoint will return the following JSON:
+```json
+{
+ "transaction-hash": "<Transaction hash that includes the delivered warp message>"
+}
+```
+
+#### `/relay/message`
+- Used to manually relay a warp message. The body of the request must contain the following JSON:
+```json
+{
+ "unsigned-message-bytes": "<Hex encoded byte array containing the unsigned warp message>",
+ "source-address": "<Hex encoding of address that sent the warp message>"
+}
+```
+- If successful, the endpoint will return the following JSON:
+```json
+{
+ "transaction-hash": "<Transaction hash that includes the delivered Warp message>",
+}
+```
+
+#### `/health`
+- Takes no arguments. Returns a `200` status code if all Application Relayers are healthy. Returns a `503` status if any of the Application Relayers have experienced an unrecoverable error. Here is an example return body:
+```json
+{
+  "status": "down",
+  "details": {
+    "relayers-all": {
+      "status": "down",
+      "timestamp": "2024-06-01T05:06:07.685522Z",
+      "error": "<List of cb-58 encoded IDs for unhealthy relayers>"
+    }
+  }
+}
+```
+
 ## Testing
 
 ### Unit Tests

api/health_check.go

@@ -0,0 +1,42 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/alexliesenfeld/health"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"go.uber.org/atomic"
+	"go.uber.org/zap"
+)
+
+const HealthAPIPath = "/health"
+
+func HandleHealthCheck(logger logging.Logger, relayerHealth map[ids.ID]*atomic.Bool) {
+	http.Handle(HealthAPIPath, healthCheckHandler(logger, relayerHealth))
+}
+
+func healthCheckHandler(logger logging.Logger, relayerHealth map[ids.ID]*atomic.Bool) http.Handler {
+	return health.NewHandler(health.NewChecker(
+		health.WithCheck(health.Check{
+			Name: "relayers-all",
+			Check: func(context.Context) error {
+				// Store the IDs as the cb58 encoding
+				var unhealthyRelayers []string
+				for id, health := range relayerHealth {
+					if !health.Load() {
+						unhealthyRelayers = append(unhealthyRelayers, id.String())
+					}
+				}
+
+				if len(unhealthyRelayers) > 0 {
+					logger.Fatal("Relayers are unhealthy for blockchains", zap.Strings("blockchains", unhealthyRelayers))
+					return fmt.Errorf("relayers are unhealthy for blockchains %v", unhealthyRelayers)
+				}
+				return nil
+			},
+		}),
+	))
+}

api/relay_message.go

@@ -0,0 +1,143 @@
+package api
+
+import (
+	"encoding/json"
+	"math/big"
+	"net/http"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/utils/logging"
+	"github.com/ava-labs/awm-relayer/relayer"
+	"github.com/ava-labs/awm-relayer/types"
+	relayerTypes "github.com/ava-labs/awm-relayer/types"
+	"github.com/ethereum/go-ethereum/common"
+	"go.uber.org/zap"
+)
+
+const (
+	RelayAPIPath        = "/relay"
+	RelayMessageAPIPath = RelayAPIPath + "/message"
+)
+
+type RelayMessageRequest struct {
+	// Required. cb58 encoding of the source blockchain ID for the message
+	BlockchainID string `json:"blockchain-id"`
+	// Required. cb58 encoding of the warp message ID
+	MessageID string `json:"message-id"`
+	// Required. Block number that the message was sent in
+	BlockNum uint64 `json:"block-num"`
+}
+
+type RelayMessageResponse struct {
+	// hex encoding of the transaction hash containing the processed message
+	TransactionHash string `json:"transaction-hash"`
+}
+
+// Defines a manual warp message to be sent from the relayer through the API.
+type ManualWarpMessageRequest struct {
+	UnsignedMessageBytes []byte `json:"unsigned-message-bytes"`
+	SourceAddress        string `json:"source-address"`
+}
+
+func HandleRelayMessage(logger logging.Logger, messageCoordinator *relayer.MessageCoordinator) {
+	http.Handle(RelayAPIPath, relayAPIHandler(logger, messageCoordinator))
+}
+
+func HandleRelay(logger logging.Logger, messageCoordinator *relayer.MessageCoordinator) {
+	http.Handle(RelayMessageAPIPath, relayMessageAPIHandler(logger, messageCoordinator))
+}
+
+func relayMessageAPIHandler(logger logging.Logger, messageCoordinator *relayer.MessageCoordinator) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req ManualWarpMessageRequest
+		err := json.NewDecoder(r.Body).Decode(&req)
+		if err != nil {
+			logger.Warn("Could not decode request body")
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		unsignedMessage, err := types.UnpackWarpMessage(req.UnsignedMessageBytes)
+		if err != nil {
+			logger.Warn("Error unpacking warp message", zap.Error(err))
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		warpMessageInfo := &relayerTypes.WarpMessageInfo{
+			SourceAddress:   common.HexToAddress(req.SourceAddress),
+			UnsignedMessage: unsignedMessage,
+		}
+
+		txHash, err := messageCoordinator.ProcessWarpMessage(warpMessageInfo)
+		if err != nil {
+			logger.Error("Error processing message", zap.Error(err))
+			http.Error(w, "error processing message: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		resp, err := json.Marshal(
+			RelayMessageResponse{
+				TransactionHash: txHash.Hex(),
+			},
+		)
+		if err != nil {
+			logger.Error("Error marshaling response", zap.Error(err))
+			http.Error(w, "error marshaling response: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		_, err = w.Write(resp)
+		if err != nil {
+			logger.Error("Error writing response", zap.Error(err))
+		}
+	})
+}
+
+func relayAPIHandler(logger logging.Logger, messageCoordinator *relayer.MessageCoordinator) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req RelayMessageRequest
+		err := json.NewDecoder(r.Body).Decode(&req)
+		if err != nil {
+			logger.Warn("Could not decode request body")
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		blockchainID, err := ids.FromString(req.BlockchainID)
+		if err != nil {
+			logger.Warn("Invalid blockchainID", zap.String("blockchainID", req.BlockchainID))
+			http.Error(w, "invalid blockchainID: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+		messageID, err := ids.FromString(req.MessageID)
+		if err != nil {
+			logger.Warn("Invalid messageID", zap.String("messageID", req.MessageID))
+			http.Error(w, "invalid messageID: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		txHash, err := messageCoordinator.ProcessMessageID(blockchainID, messageID, new(big.Int).SetUint64(req.BlockNum))
+		if err != nil {
+			logger.Error("Error processing message", zap.Error(err))
+			http.Error(w, "error processing message: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		resp, err := json.Marshal(
+			RelayMessageResponse{
+				TransactionHash: txHash.Hex(),
+			},
+		)
+		if err != nil {
+			logger.Error("Error marshalling response", zap.Error(err))
+			http.Error(w, "error marshalling response: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		_, err = w.Write(resp)
+		if err != nil {
+			logger.Error("Error writing response", zap.Error(err))
+		}
+	})
+}

config/config.go

@@ -58,7 +58,6 @@ type Config struct {
 	SourceBlockchains      []*SourceBlockchain      `mapstructure:"source-blockchains" json:"source-blockchains"`
 	DestinationBlockchains []*DestinationBlockchain `mapstructure:"destination-blockchains" json:"destination-blockchains"`
 	ProcessMissedBlocks    bool                     `mapstructure:"process-missed-blocks" json:"process-missed-blocks"`
-	ManualWarpMessages     []*ManualWarpMessage     `mapstructure:"manual-warp-messages" json:"manual-warp-messages"`
 
 	// convenience field to fetch a blockchain's subnet ID
 	blockchainIDToSubnetID map[ids.ID]ids.ID
@@ -71,7 +70,7 @@ func DisplayUsageText() {
 
 // Validates the configuration
 // Does not modify the public fields as derived from the configuration passed to the application,
-// but does initialize private fields available through getters
+// but does initialize private fields available through getters.
 func (c *Config) Validate() error {
 	if len(c.SourceBlockchains) == 0 {
 		return errors.New("relayer not configured to relay from any subnets. A list of source subnets must be provided in the configuration file")
@@ -120,13 +119,6 @@ func (c *Config) Validate() error {
 	}
 	c.blockchainIDToSubnetID = blockchainIDToSubnetID
 
-	// Validate the manual warp messages
-	for i, msg := range c.ManualWarpMessages {
-		if err := msg.Validate(); err != nil {
-			return fmt.Errorf("invalid manual warp message at index %d: %w", i, err)
-		}
-	}
-
 	return nil
 }