Node: Channel writes without blocking

Author: bruce-riley

node/cmd/spy/spy.go

@@ -123,7 +123,7 @@ func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error {
 					return err
 				}
 			}
-			sub.ch <- message{vaaBytes: vaaBytes}
+			sub.ch <- message{vaaBytes: vaaBytes} //can_block: Don't want to drop incoming VAAs
 			continue
 		}
 
@@ -143,7 +143,7 @@ func (s *spyServer) PublishSignedVAA(vaaBytes []byte) error {
 						return err
 					}
 				}
-				sub.ch <- message{vaaBytes: vaaBytes}
+				sub.ch <- message{vaaBytes: vaaBytes} //can_block: Don't want to drop incoming VAAs
 			}
 		}
 
@@ -246,7 +246,7 @@ func newSpyServer(logger *zap.Logger) *spyServer {
 func DoWithTimeout(f func() error, d time.Duration) error {
 	errChan := make(chan error, 1)
 	go func() {
-		errChan <- f()
+		errChan <- f() //can_block: Has timeout below
 		close(errChan)
 	}()
 	t := time.NewTimer(d)

node/hack/evm_test/wstest.go

@@ -77,7 +77,7 @@ func main() {
 				case <-ctx.Done():
 					return
 				case err := <-headerSubscription.Err():
-					errC <- fmt.Errorf("block subscription failed: %w", err)
+					errC <- fmt.Errorf("block subscription failed: %w", err) //can_block: Only does one write
 					return
 				case block := <-headSink:
 					// These two pointers should have been checked before the event was placed on the channel, but just being safe.
@@ -114,7 +114,7 @@ func main() {
 				case <-ctx.Done():
 					return
 				case err := <-messageSub.Err():
-					errC <- fmt.Errorf("message subscription failed: %w", err)
+					errC <- fmt.Errorf("message subscription failed: %w", err) //can_block: Only does one write
 					return
 				case ev := <-messageC:
 					logger.Info("Received a log event from the contract", zap.Any("ev", ev))

node/pkg/adminrpc/adminserver.go

@@ -797,7 +797,7 @@ func (s *nodePrivilegedService) InjectGovernanceVAA(ctx context.Context, req *no
 
 		vaaInjectionsTotal.Inc()
 
-		s.injectC <- &common.MessagePublication{
+		s.injectC <- &common.MessagePublication{ //can_block: Only blocks this command
 			TxID:             ethcommon.Hash{}.Bytes(),
 			Timestamp:        v.Timestamp,
 			Nonce:            v.Nonce,
@@ -897,7 +897,7 @@ func (s *nodePrivilegedService) fetchMissing(
 			// Inject into the gossip signed VAA receive path.
 			// This has the same effect as if the VAA was received from the network
 			// (verifying signature, storing in local DB...).
-			s.signedInC <- &gossipv1.SignedVAAWithQuorum{
+			s.signedInC <- &gossipv1.SignedVAAWithQuorum{ //can_block: Only blocks this command
 				Vaa: vaaBytes,
 			}
 

node/pkg/common/channel_utils.go

@@ -2,8 +2,29 @@ package common
 
 import (
 	"context"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	channelWriteDrops = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "wormhole_channel_write_drops",
+			Help: "Total number of channel writes that were dropped due to channel overflow",
+		}, []string{"channel_id"})
 )
 
+// WriteToChannelWithoutBlocking attempts to write the specified event to the specified channel. If the write would block,
+// it increments the `channelWriteDrops` metric with the specified channel ID.
+func WriteToChannelWithoutBlocking[T any](channel chan<- T, evt T, label string) {
+	select {
+	case channel <- evt:
+	default:
+		channelWriteDrops.WithLabelValues(label).Inc()
+	}
+}
+
 // ReadFromChannelWithTimeout reads events from the channel until a timeout occurs or the max maxCount is reached.
 func ReadFromChannelWithTimeout[T any](ctx context.Context, ch <-chan T, maxCount int) ([]T, error) {
 	out := make([]T, 0, maxCount)

node/pkg/common/channel_utils_test.go

@@ -78,3 +78,17 @@ func TestReadFromChannelWithTimeout_TooMuchData(t *testing.T) {
 	require.Equal(t, 1, len(observations))
 	assert.Equal(t, 3, observations[0])
 }
+
+func TestWriteToChannelWithoutBlocking(t *testing.T) {
+	myChan := make(chan int, 1)
+	assert.Equal(t, 0.0, getCounterValue(channelWriteDrops, "numbers"))
+	WriteToChannelWithoutBlocking(myChan, 42, "numbers")
+	assert.Equal(t, 0.0, getCounterValue(channelWriteDrops, "numbers"))
+	WriteToChannelWithoutBlocking(myChan, 43, "numbers")
+	assert.Equal(t, 1.0, getCounterValue(channelWriteDrops, "numbers"))
+	WriteToChannelWithoutBlocking(myChan, 44, "numbers")
+	assert.Equal(t, 2.0, getCounterValue(channelWriteDrops, "numbers"))
+	WriteToChannelWithoutBlocking(myChan, 44, "different_label")
+	assert.Equal(t, 1.0, getCounterValue(channelWriteDrops, "different_label"))
+	assert.Equal(t, 2.0, getCounterValue(channelWriteDrops, "numbers"))
+}

node/pkg/common/guardianset.go

@@ -177,7 +177,7 @@ func (st *GuardianSetState) SetHeartbeat(addr common.Address, peerId peer.ID, hb
 
 	v[peerId] = hb
 	if st.updateC != nil {
-		st.updateC <- hb
+		WriteToChannelWithoutBlocking(st.updateC, hb, "heartbeat")
 	}
 	return nil
 }

node/pkg/governor/governor_monitoring.go

@@ -80,6 +80,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/certusone/wormhole/node/pkg/guardiansigner"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
 	publicrpcv1 "github.com/certusone/wormhole/node/pkg/proto/publicrpc/v1"
@@ -618,7 +619,7 @@ func (gov *ChainGovernor) publishConfig(ctx context.Context, hb *gossipv1.Heartb
 		panic(err)
 	}
 
-	sendC <- b
+	common.WriteToChannelWithoutBlocking(sendC, b, "gov_config_gossip_out")
 }
 
 func (gov *ChainGovernor) publishStatus(ctx context.Context, hb *gossipv1.Heartbeat, sendC chan<- []byte, startTime time.Time, guardianSigner guardiansigner.GuardianSigner, ourAddr ethCommon.Address) {
@@ -703,5 +704,5 @@ func (gov *ChainGovernor) publishStatus(ctx context.Context, hb *gossipv1.Heartb
 		panic(err)
 	}
 
-	sendC <- b
+	common.WriteToChannelWithoutBlocking(sendC, b, "gov_status_gossip_out")
 }

node/pkg/governor/governor_prices.go

@@ -158,7 +158,7 @@ func (gov *ChainGovernor) queryCoinGecko(ctx context.Context) error {
 		for {
 			select {
 			case <-ticker.C:
-				throttle <- 1
+				throttle <- 1 //can_block: We want this to block for throttling
 			case <-ctx.Done():
 				return
 			}

node/pkg/node/node.go

@@ -34,6 +34,10 @@ const (
 	// per second during normal operations. However, since some messages get published immediately, we need to allow extra room.
 	inboundBatchObservationBufferSize = 1000
 
+	// inboundMessageBufferSize configures the size of the msgC channel used to publish new observations from the watcher to the processor.
+	// This channel is shared across all the watchers so we don't want to hang up other watchers while the processor is handling an observation from one.
+	inboundMessageBufferSize = 1000
+
 	// inboundSignedVaaBufferSize configures the size of the signedInC channel that contains VAAs from other Guardians.
 	// One VAA takes roughly 0.01ms to process if we already have one in the database and 2ms if we don't.
 	// So in the worst case the entire queue can be processed in 2s.
@@ -122,7 +126,7 @@ func (g *G) initializeBasic(rootCtxCancel context.CancelFunc) {
 	g.gossipAttestationSendC = make(chan []byte, gossipAttestationSendBufferSize)
 	g.gossipVaaSendC = make(chan []byte, gossipVaaSendBufferSize)
 	g.batchObsvC = makeChannelPair[*common.MsgWithTimeStamp[gossipv1.SignedObservationBatch]](inboundBatchObservationBufferSize)
-	g.msgC = makeChannelPair[*common.MessagePublication](0)
+	g.msgC = makeChannelPair[*common.MessagePublication](inboundMessageBufferSize)
 	g.setC = makeChannelPair[*common.GuardianSet](1) // This needs to be a buffered channel because of a circular dependency between processor and accountant during startup.
 	g.signedInC = makeChannelPair[*gossipv1.SignedVAAWithQuorum](inboundSignedVaaBufferSize)
 	g.obsvReqC = makeChannelPair[*gossipv1.ObservationRequest](observationRequestInboundBufferSize)

node/pkg/node/options.go

@@ -382,7 +382,7 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 									zap.String("txID", msg.TxIDString()),
 									zap.Time("timestamp", msg.Timestamp))
 							} else {
-								g.msgC.writeC <- msg
+								g.msgC.writeC <- msg //can_block: The channel to the processor is buffered and shared across chains, if it backs up we should stop processing new observations
 							}
 						}
 					}
@@ -408,7 +408,7 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 									zap.Stringer("watcherChainId", chainId),
 								)
 							}
-							g.queryResponseC.writeC <- response
+							g.queryResponseC.writeC <- response //can_block: This channel is buffered, if it backs up we'll stop processing queries until it clears
 						}
 					}
 				}(chainQueryResponseC[chainId], chainId)