Node: Reobserve with custom endpoint (#4260)

* Node: Reobserve with custom endpoint

* Code review rework

* Node/Solana: Code review rework

* Code review rework

* Verify that we are connecting to the correct chain

Author: bruce-riley

node/cmd/guardiand/adminclient.go

@@ -23,6 +23,7 @@ import (
 	"golang.org/x/crypto/sha3"
 
 	"github.com/certusone/wormhole/node/pkg/guardiansigner"
+	"github.com/certusone/wormhole/node/pkg/node"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
 	publicrpcv1 "github.com/certusone/wormhole/node/pkg/proto/publicrpc/v1"
 	"github.com/wormhole-foundation/wormhole/sdk"
@@ -62,6 +63,7 @@ func init() {
 	DumpVAAByMessageID.Flags().AddFlagSet(pf)
 	DumpRPCs.Flags().AddFlagSet(pf)
 	SendObservationRequest.Flags().AddFlagSet(pf)
+	ReobserveWithEndpoint.Flags().AddFlagSet(pf)
 	ClientChainGovernorStatusCmd.Flags().AddFlagSet(pf)
 	ClientChainGovernorReloadCmd.Flags().AddFlagSet(pf)
 	ClientChainGovernorDropPendingVAACmd.Flags().AddFlagSet(pf)
@@ -84,6 +86,7 @@ func init() {
 	AdminCmd.AddCommand(DumpVAAByMessageID)
 	AdminCmd.AddCommand(DumpRPCs)
 	AdminCmd.AddCommand(SendObservationRequest)
+	AdminCmd.AddCommand(ReobserveWithEndpoint)
 	AdminCmd.AddCommand(ClientChainGovernorStatusCmd)
 	AdminCmd.AddCommand(ClientChainGovernorReloadCmd)
 	AdminCmd.AddCommand(ClientChainGovernorDropPendingVAACmd)
@@ -136,6 +139,13 @@ var SendObservationRequest = &cobra.Command{
 	Args:  cobra.ExactArgs(2),
 }
 
+var ReobserveWithEndpoint = &cobra.Command{
+	Use:   "reobserve-with-endpoint [CHAIN_ID|CHAIN_NAME] [TX_HASH_HEX] [CUSTOM_URL]",
+	Short: "Performs a local reobservation for the given chain ID and chain-specific tx_hash using the specified endpoint",
+	Run:   runReobserveWithEndpoint,
+	Args:  cobra.ExactArgs(3),
+}
+
 var ClientChainGovernorStatusCmd = &cobra.Command{
 	Use:   "governor-status",
 	Short: "Displays the status of the chain governor",
@@ -415,6 +425,51 @@ func runSendObservationRequest(cmd *cobra.Command, args []string) {
 	}
 }
 
+func runReobserveWithEndpoint(cmd *cobra.Command, args []string) {
+	chainID, err := parseChainID(args[0])
+	if err != nil {
+		log.Fatalf("invalid chain ID: %v", err)
+	}
+
+	// Support tx with or without leading 0x.
+	txHash, err := hex.DecodeString(strings.TrimPrefix(args[1], "0x"))
+	if err != nil {
+		txHash, err = base58.Decode(args[1])
+		if err != nil {
+			log.Fatalf("invalid transaction hash (neither hex nor base58): %v", err)
+		}
+	}
+
+	url := args[2]
+	if valid := node.ValidateURL(url, []string{"http", "https"}); !valid {
+		log.Fatalf(`invalid url, must be "http" or "https"`)
+	}
+
+	// Allow extra time since the watcher can block on the reobservation.
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	conn, c, err := getAdminClient(ctx, *clientSocketPath)
+	if err != nil {
+		log.Fatalf("failed to get admin client: %v", err)
+	}
+	defer conn.Close()
+
+	resp, err := c.ReobserveWithEndpoint(ctx, &nodev1.ReobserveWithEndpointRequest{
+		ChainId: uint32(chainID),
+		TxHash:  txHash,
+		Url:     url,
+	})
+	if err != nil {
+		log.Fatalf("failed to send observation request with endpoint: %v", err)
+	}
+	if resp.NumObservations == 0 {
+		fmt.Println("Did not reobserve anything")
+	} else {
+		fmt.Println("Reobserved", resp.NumObservations, "messages")
+	}
+}
+
 func runDumpRPCs(cmd *cobra.Command, args []string) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()

node/pkg/adminrpc/adminserver.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/certusone/wormhole/node/pkg/guardiansigner"
 	"github.com/certusone/wormhole/node/pkg/watchers/evm/connectors"
+	"github.com/certusone/wormhole/node/pkg/watchers/interfaces"
 	"github.com/holiman/uint256"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -65,6 +66,7 @@ type nodePrivilegedService struct {
 	guardianSigner  guardiansigner.GuardianSigner
 	guardianAddress ethcommon.Address
 	rpcMap          map[string]string
+	reobservers     interfaces.Reobservers
 }
 
 func NewPrivService(
@@ -78,6 +80,7 @@ func NewPrivService(
 	guardianSigner guardiansigner.GuardianSigner,
 	guardianAddress ethcommon.Address,
 	rpcMap map[string]string,
+	reobservers interfaces.Reobservers,
 
 ) *nodePrivilegedService {
 	return &nodePrivilegedService{
@@ -91,6 +94,7 @@ func NewPrivService(
 		guardianSigner:  guardianSigner,
 		guardianAddress: guardianAddress,
 		rpcMap:          rpcMap,
+		reobservers:     reobservers,
 	}
 }
 
@@ -962,6 +966,20 @@ func (s *nodePrivilegedService) SendObservationRequest(ctx context.Context, req
 	return &nodev1.SendObservationRequestResponse{}, nil
 }
 
+func (s *nodePrivilegedService) ReobserveWithEndpoint(ctx context.Context, req *nodev1.ReobserveWithEndpointRequest) (*nodev1.ReobserveWithEndpointResponse, error) {
+	watcher := s.reobservers[vaa.ChainID(req.ChainId)]
+	if watcher == nil {
+		return nil, status.Errorf(codes.Internal, "chain %d does not support reobservation by endpoint", req.ChainId)
+	}
+
+	numObservations, err := watcher.Reobserve(ctx, vaa.ChainID(req.ChainId), req.TxHash, req.Url)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "reobservation failed: %v", err)
+	}
+
+	return &nodev1.ReobserveWithEndpointResponse{NumObservations: numObservations}, nil
+}
+
 func (s *nodePrivilegedService) ChainGovernorStatus(ctx context.Context, req *nodev1.ChainGovernorStatusRequest) (*nodev1.ChainGovernorStatusResponse, error) {
 	if s.governor == nil {
 		return nil, fmt.Errorf("chain governor is not enabled")

node/pkg/node/adminServiceRunnable.go

@@ -18,6 +18,7 @@ import (
 	"github.com/certusone/wormhole/node/pkg/publicrpc"
 	"github.com/certusone/wormhole/node/pkg/supervisor"
 	"github.com/certusone/wormhole/node/pkg/watchers/evm/connectors"
+	"github.com/certusone/wormhole/node/pkg/watchers/interfaces"
 	"go.uber.org/zap"
 
 	ethcommon "github.com/ethereum/go-ethereum/common"
@@ -37,6 +38,7 @@ func adminServiceRunnable(
 	ethRpc *string,
 	ethContract *string,
 	rpcMap map[string]string,
+	reobservers interfaces.Reobservers,
 ) (supervisor.Runnable, error) {
 	// Delete existing UNIX socket, if present.
 	fi, err := os.Stat(socketPath)
@@ -92,6 +94,7 @@ func adminServiceRunnable(
 		guardianSigner,
 		ethcrypto.PubkeyToAddress(guardianSigner.PublicKey(ctx)),
 		rpcMap,
+		reobservers,
 	)
 
 	publicrpcService := publicrpc.NewPublicrpcServer(logger, db, gst, gov)

node/pkg/node/node.go

@@ -13,6 +13,7 @@ import (
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
 	"github.com/certusone/wormhole/node/pkg/query"
 	"github.com/certusone/wormhole/node/pkg/supervisor"
+	"github.com/certusone/wormhole/node/pkg/watchers/interfaces"
 	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 
 	"go.uber.org/zap"
@@ -74,6 +75,7 @@ type G struct {
 	// runnables
 	runnablesWithScissors map[string]supervisor.Runnable
 	runnables             map[string]supervisor.Runnable
+	reobservers           interfaces.Reobservers
 
 	// various channels
 	// Outbound gossip message queues (need to be read/write because p2p needs read/write)
@@ -140,6 +142,7 @@ func (g *G) initializeBasic(rootCtxCancel context.CancelFunc) {
 	// allocate maps
 	g.runnablesWithScissors = make(map[string]supervisor.Runnable)
 	g.runnables = make(map[string]supervisor.Runnable)
+	g.reobservers = make(interfaces.Reobservers)
 }
 
 // applyOptions applies `options` to the GuardianNode.

node/pkg/node/options.go

@@ -440,14 +440,18 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 					wc.SetL1Finalizer(l1watcher)
 				}
 
-				l1finalizer, runnable, err := wc.Create(chainMsgC[wc.GetChainID()], chainObsvReqC[wc.GetChainID()], g.chainQueryReqC[wc.GetChainID()], chainQueryResponseC[wc.GetChainID()], g.setC.writeC, g.env)
+				l1finalizer, runnable, reobserver, err := wc.Create(chainMsgC[wc.GetChainID()], chainObsvReqC[wc.GetChainID()], g.chainQueryReqC[wc.GetChainID()], chainQueryResponseC[wc.GetChainID()], g.setC.writeC, g.env)
 
 				if err != nil {
 					return fmt.Errorf("error creating watcher: %w", err)
 				}
 
 				g.runnablesWithScissors[watcherName] = runnable
 				watchers[wc.GetNetworkID()] = l1finalizer
+
+				if reobserver != nil {
+					g.reobservers[wc.GetChainID()] = reobserver
+				}
 			}
 
 			if ibcWatcherConfig != nil {
@@ -509,6 +513,7 @@ func GuardianOptionAdminService(socketPath string, ethRpc *string, ethContract *
 				ethRpc,
 				ethContract,
 				rpcMap,
+				g.reobservers,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to create admin service: %w", err)

node/pkg/node/url_verification.go

@@ -20,7 +20,7 @@ func hasKnownSchemePrefix(urlStr string) bool {
 	return false
 }
 
-func validateURL(urlStr string, validSchemes []string) bool {
+func ValidateURL(urlStr string, validSchemes []string) bool {
 	// If no scheme is required, validate host:port format
 	if len(validSchemes) == 1 && validSchemes[0] == "" {
 		host, port, err := net.SplitHostPort(urlStr)
@@ -69,7 +69,7 @@ func RegisterFlagWithValidationOrFail(cmd *cobra.Command, name string, descripti
 			return
 		}
 
-		if valid := validateURL(*flagValue, expectedSchemes); !valid {
+		if valid := ValidateURL(*flagValue, expectedSchemes); !valid {
 			log.Fatalf("Invalid format for flag --%s. Expected format: %s. Example: '%s'", name, formatExample, example)
 		}
 	})

node/pkg/node/url_verification_test.go

@@ -29,7 +29,7 @@ func TestValidateURL(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		result := validateURL(test.urlStr, test.validSchemes)
+		result := ValidateURL(test.urlStr, test.validSchemes)
 		assert.Equal(t, test.expected, result)
 	}
 }