diff --git a/node/cmd/guardiand/node.go b/node/cmd/guardiand/node.go index 3197c81f9..1b752134c 100644 --- a/node/cmd/guardiand/node.go +++ b/node/cmd/guardiand/node.go @@ -1097,7 +1097,7 @@ func runNode(cmd *cobra.Command, args []string) { readiness.RegisterComponent(common.ReadinessTerraSyncing) chainObsvReqC[vaa.ChainIDTerra] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize) if err := supervisor.Run(ctx, "terrawatch", - cosmwasm.NewWatcher(*terraWS, *terraLCD, *terraContract, lockC, chainObsvReqC[vaa.ChainIDTerra], common.ReadinessTerraSyncing, vaa.ChainIDTerra).Run); err != nil { + common.WrapWithScissors(cosmwasm.NewWatcher(*terraWS, *terraLCD, *terraContract, lockC, chainObsvReqC[vaa.ChainIDTerra], common.ReadinessTerraSyncing, vaa.ChainIDTerra).Run, "terrawatch")); err != nil { return err } } @@ -1107,7 +1107,7 @@ func runNode(cmd *cobra.Command, args []string) { readiness.RegisterComponent(common.ReadinessTerra2Syncing) chainObsvReqC[vaa.ChainIDTerra2] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize) if err := supervisor.Run(ctx, "terra2watch", - cosmwasm.NewWatcher(*terra2WS, *terra2LCD, *terra2Contract, lockC, chainObsvReqC[vaa.ChainIDTerra2], common.ReadinessTerra2Syncing, vaa.ChainIDTerra2).Run); err != nil { + common.WrapWithScissors(cosmwasm.NewWatcher(*terra2WS, *terra2LCD, *terra2Contract, lockC, chainObsvReqC[vaa.ChainIDTerra2], common.ReadinessTerra2Syncing, vaa.ChainIDTerra2).Run, "terra2watch")); err != nil { return err } } @@ -1117,7 +1117,7 @@ func runNode(cmd *cobra.Command, args []string) { readiness.RegisterComponent(common.ReadinessXplaSyncing) chainObsvReqC[vaa.ChainIDXpla] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize) if err := supervisor.Run(ctx, "xplawatch", - cosmwasm.NewWatcher(*xplaWS, *xplaLCD, *xplaContract, lockC, chainObsvReqC[vaa.ChainIDXpla], common.ReadinessXplaSyncing, vaa.ChainIDXpla).Run); err != nil { + common.WrapWithScissors(cosmwasm.NewWatcher(*xplaWS, *xplaLCD, *xplaContract, lockC, chainObsvReqC[vaa.ChainIDXpla], common.ReadinessXplaSyncing, vaa.ChainIDXpla).Run, "xplawatch")); err != nil { return err } } @@ -1205,7 +1205,7 @@ func runNode(cmd *cobra.Command, args []string) { readiness.RegisterComponent(common.ReadinessInjectiveSyncing) chainObsvReqC[vaa.ChainIDInjective] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize) if err := supervisor.Run(ctx, "injectivewatch", - cosmwasm.NewWatcher(*injectiveWS, *injectiveLCD, *injectiveContract, lockC, chainObsvReqC[vaa.ChainIDInjective], common.ReadinessInjectiveSyncing, vaa.ChainIDInjective).Run); err != nil { + common.WrapWithScissors(cosmwasm.NewWatcher(*injectiveWS, *injectiveLCD, *injectiveContract, lockC, chainObsvReqC[vaa.ChainIDInjective], common.ReadinessInjectiveSyncing, vaa.ChainIDInjective).Run, "injectivewatch")); err != nil { return err } } diff --git a/node/pkg/watchers/cosmwasm/watcher.go b/node/pkg/watchers/cosmwasm/watcher.go index d725a5d2e..18faae25d 100644 --- a/node/pkg/watchers/cosmwasm/watcher.go +++ b/node/pkg/watchers/cosmwasm/watcher.go @@ -21,10 +21,12 @@ import ( "github.com/certusone/wormhole/node/pkg/common" "github.com/certusone/wormhole/node/pkg/readiness" "github.com/certusone/wormhole/node/pkg/supervisor" - "github.com/gorilla/websocket" + "github.com/tidwall/gjson" "github.com/wormhole-foundation/wormhole/sdk/vaa" "go.uber.org/zap" + "nhooyr.io/websocket" + "nhooyr.io/websocket/wsjson" ) type ( @@ -141,13 +143,18 @@ func (e *Watcher) Run(ctx context.Context) error { logger.Info("connecting to websocket", zap.String("network", networkName), zap.String("url", e.urlWS)) - c, _, err := websocket.DefaultDialer.DialContext(ctx, e.urlWS, nil) + c, _, err := websocket.Dial(ctx, e.urlWS, nil) if err != nil { p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) connectionErrors.WithLabelValues(networkName, "websocket_dial_error").Inc() return fmt.Errorf("websocket dial failed: %w", err) } - defer c.Close() + defer c.Close(websocket.StatusNormalClosure, "") + + // During testing, I got a message larger then the default + // 32768. Increasing this limit effects an internal buffer that is used + // to as part of the zero alloc/copy design. + c.SetReadLimit(524288) // Subscribe to smart contract transactions params := [...]string{fmt.Sprintf("tm.event='Tx' AND %s='%s'", e.contractAddressFilterKey, e.contract)} @@ -157,7 +164,7 @@ func (e *Watcher) Run(ctx context.Context) error { Params: params, ID: 1, } - err = c.WriteJSON(command) + err = wsjson.Write(ctx, c, command) if err != nil { p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) connectionErrors.WithLabelValues(networkName, "websocket_subscription_error").Inc() @@ -165,7 +172,7 @@ func (e *Watcher) Run(ctx context.Context) error { } // Wait for the success response - _, _, err = c.ReadMessage() + _, _, err = c.Read(ctx) if err != nil { p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) connectionErrors.WithLabelValues(networkName, "event_subscription_error").Inc() @@ -175,49 +182,53 @@ func (e *Watcher) Run(ctx context.Context) error { readiness.SetReady(e.readiness) - go func() { + common.RunWithScissors(ctx, errC, "cosmwasm_block_height", func(ctx context.Context) error { t := time.NewTicker(5 * time.Second) client := &http.Client{ Timeout: time.Second * 5, } for { - <-t.C - msm := time.Now() - // Query and report height and set currentSlotHeight - resp, err := client.Get(fmt.Sprintf("%s/%s", e.urlLCD, e.latestBlockURL)) - if err != nil { - logger.Error("query latest block response error", zap.String("network", networkName), zap.Error(err)) - continue - } - blocksBody, err := io.ReadAll(resp.Body) - if err != nil { - logger.Error("query latest block response read error", zap.String("network", networkName), zap.Error(err)) - errC <- err + select { + case <-ctx.Done(): + return nil + case <-t.C: + msm := time.Now() + // Query and report height and set currentSlotHeight + resp, err := client.Get(fmt.Sprintf("%s/%s", e.urlLCD, e.latestBlockURL)) + if err != nil { + logger.Error("query latest block response error", zap.String("network", networkName), zap.Error(err)) + continue + } + blocksBody, err := io.ReadAll(resp.Body) + if err != nil { + logger.Error("query latest block response read error", zap.String("network", networkName), zap.Error(err)) + errC <- err + resp.Body.Close() + continue + } resp.Body.Close() - continue + + // Update the prom metrics with how long the http request took to the rpc + queryLatency.WithLabelValues(networkName, "block_latest").Observe(time.Since(msm).Seconds()) + + blockJSON := string(blocksBody) + latestBlock := gjson.Get(blockJSON, "block.header.height") + logger.Info("current height", zap.String("network", networkName), zap.Int64("block", latestBlock.Int())) + currentSlotHeight.WithLabelValues(networkName).Set(float64(latestBlock.Int())) + p2p.DefaultRegistry.SetNetworkStats(e.chainID, &gossipv1.Heartbeat_Network{ + Height: latestBlock.Int(), + ContractAddress: e.contract, + }) } - resp.Body.Close() - - // Update the prom metrics with how long the http request took to the rpc - queryLatency.WithLabelValues(networkName, "block_latest").Observe(time.Since(msm).Seconds()) - - blockJSON := string(blocksBody) - latestBlock := gjson.Get(blockJSON, "block.header.height") - logger.Info("current height", zap.String("network", networkName), zap.Int64("block", latestBlock.Int())) - currentSlotHeight.WithLabelValues(networkName).Set(float64(latestBlock.Int())) - p2p.DefaultRegistry.SetNetworkStats(e.chainID, &gossipv1.Heartbeat_Network{ - Height: latestBlock.Int(), - ContractAddress: e.contract, - }) } - }() + }) - go func() { + common.RunWithScissors(ctx, errC, "cosmwasm_objs_req", func(ctx context.Context) error { for { select { case <-ctx.Done(): - return + return nil case r := <-e.obsvReqC: if vaa.ChainID(r.ChainId) != e.chainID { panic("invalid chain ID") @@ -267,53 +278,52 @@ func (e *Watcher) Run(ctx context.Context) error { } } } - }() - - go func() { - defer close(errC) + }) + common.RunWithScissors(ctx, errC, "cosmwasm_data_pump", func(ctx context.Context) error { for { - _, message, err := c.ReadMessage() - if err != nil { - p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) - connectionErrors.WithLabelValues(networkName, "channel_read_error").Inc() - logger.Error("error reading channel", zap.String("network", networkName), zap.Error(err)) - errC <- err - return + select { + case <-ctx.Done(): + return nil + default: + _, message, err := c.Read(ctx) + if err != nil { + p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) + connectionErrors.WithLabelValues(networkName, "channel_read_error").Inc() + logger.Error("error reading channel", zap.String("network", networkName), zap.Error(err)) + errC <- err + return nil + } + + // Received a message from the blockchain + json := string(message) + + txHashRaw := gjson.Get(json, "result.events.tx\\.hash.0") + if !txHashRaw.Exists() { + logger.Warn("message does not have tx hash", zap.String("network", networkName), zap.String("payload", json)) + continue + } + txHash := txHashRaw.String() + + events := gjson.Get(json, "result.data.value.TxResult.result.events") + if !events.Exists() { + logger.Warn("message has no events", zap.String("network", networkName), zap.String("payload", json)) + continue + } + + msgs := EventsToMessagePublications(e.contract, txHash, events.Array(), logger, e.chainID, e.contractAddressLogKey) + for _, msg := range msgs { + e.msgChan <- msg + messagesConfirmed.WithLabelValues(networkName).Inc() + } + + // We do not send guardian changes to the processor - ETH guardians are the source of truth. } - - // Received a message from the blockchain - json := string(message) - - txHashRaw := gjson.Get(json, "result.events.tx\\.hash.0") - if !txHashRaw.Exists() { - logger.Warn("message does not have tx hash", zap.String("network", networkName), zap.String("payload", json)) - continue - } - txHash := txHashRaw.String() - - events := gjson.Get(json, "result.data.value.TxResult.result.events") - if !events.Exists() { - logger.Warn("message has no events", zap.String("network", networkName), zap.String("payload", json)) - continue - } - - msgs := EventsToMessagePublications(e.contract, txHash, events.Array(), logger, e.chainID, e.contractAddressLogKey) - for _, msg := range msgs { - e.msgChan <- msg - messagesConfirmed.WithLabelValues(networkName).Inc() - } - - // We do not send guardian changes to the processor - ETH guardians are the source of truth. } - }() + }) select { case <-ctx.Done(): - err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) - if err != nil { - logger.Error("error on closing socket ", zap.String("network", networkName), zap.Error(err)) - } return ctx.Err() case err := <-errC: return err