Reobservation rate-limit

node/cmd/guardiand/node.go

@@ -319,6 +319,9 @@ var NodeCmd = &cobra.Command{
 // guardians to reduce risk from a compromised builder.
 var Build = "prod"
 
+// observationRequestBufferSize is the buffer size of the per-network reobservation channel
+const observationRequestBufferSize = 25
+
 func runNode(cmd *cobra.Command, args []string) {
 	if Build == "dev" && !*unsafeDevMode {
 		fmt.Println("This is a development build. --unsafeDevMode must be enabled.")
@@ -790,30 +793,30 @@ func runNode(cmd *cobra.Command, args []string) {
 	chainObsvReqC := make(map[vaa.ChainID]chan *gossipv1.ObservationRequest)
 
 	// Observation request channel for each chain supporting observation requests.
-	chainObsvReqC[vaa.ChainIDSolana] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDEthereum] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDTerra] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDTerra2] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDBSC] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDPolygon] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDAvalanche] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDOasis] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDAlgorand] = make(chan *gossipv1.ObservationRequest)
+	chainObsvReqC[vaa.ChainIDSolana] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDEthereum] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDTerra] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDTerra2] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDBSC] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDPolygon] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDAvalanche] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDOasis] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDAlgorand] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
 	if *nearRPC != "" {
-		chainObsvReqC[vaa.ChainIDNear] = make(chan *gossipv1.ObservationRequest)
-	}
-	chainObsvReqC[vaa.ChainIDAurora] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDFantom] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDKarura] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDAcala] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDKlaytn] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDCelo] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDPythNet] = make(chan *gossipv1.ObservationRequest)
-	chainObsvReqC[vaa.ChainIDMoonbeam] = make(chan *gossipv1.ObservationRequest)
+		chainObsvReqC[vaa.ChainIDNear] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	}
+	chainObsvReqC[vaa.ChainIDAurora] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDFantom] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDKarura] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDAcala] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDKlaytn] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDCelo] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDPythNet] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+	chainObsvReqC[vaa.ChainIDMoonbeam] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
 	if *testnetMode {
-		chainObsvReqC[vaa.ChainIDNeon] = make(chan *gossipv1.ObservationRequest)
-		chainObsvReqC[vaa.ChainIDEthereumRopsten] = make(chan *gossipv1.ObservationRequest)
-		chainObsvReqC[vaa.ChainIDInjective] = make(chan *gossipv1.ObservationRequest)
+		chainObsvReqC[vaa.ChainIDNeon] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+		chainObsvReqC[vaa.ChainIDEthereumRopsten] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
+		chainObsvReqC[vaa.ChainIDInjective] = make(chan *gossipv1.ObservationRequest, observationRequestBufferSize)
 	}
 	go handleReobservationRequests(rootCtx, clock.New(), logger, obsvReqC, chainObsvReqC)
 

node/cmd/guardiand/reobserve.go

@@ -57,10 +57,16 @@ func handleReobservationRequests(
 				continue
 			}
 
-			cache[r] = clock.Now()
-
 			if channel, ok := chainObsvReqC[r.chainId]; ok {
-				channel <- req
+				select {
+				case channel <- req:
+					cache[r] = clock.Now()
+
+				default:
+					logger.Warn("failed to send reobservation request to watcher",
+						zap.Stringer("chain_id", r.chainId),
+						zap.String("tx_hash", r.txHash))
+				}
 			} else {
 				logger.Error("unknown chain ID for reobservation request",
 					zap.Uint16("chain_id", uint16(r.chainId)),

node/cmd/guardiand/reobserve_test.go

@@ -29,7 +29,7 @@ func setUpReobservationTest() (reobservationTestContext, func()) {
 
 	chainObsvReqC := make(map[vaa.ChainID]chan *gossipv1.ObservationRequest)
 	for i := 0; i < 10; i++ {
-		chainObsvReqC[vaa.ChainID(i)] = make(chan *gossipv1.ObservationRequest)
+		chainObsvReqC[vaa.ChainID(i)] = make(chan *gossipv1.ObservationRequest, 1)
 	}
 
 	go handleReobservationRequests(ctx, clock, zap.NewNop(), obsvReqC, chainObsvReqC)
@@ -177,3 +177,36 @@ func TestReobservationCacheEviction(t *testing.T) {
 	require.True(t, ok)
 	assert.Equal(t, req, actual)
 }
+
+func TestBlockingSend(t *testing.T) {
+	ctx, cancel := setUpReobservationTest()
+	defer cancel()
+
+	req := &gossipv1.ObservationRequest{
+		ChainId: 1,
+		TxHash:  []byte{0xe5, 0x9c, 0x1b, 0xe5, 0x0b, 0xe7, 0xe4, 0x7e},
+	}
+
+	// Send one reobservation request but don't drain it from the chain-specific channel.
+	ctx.obsvReqC <- req
+
+	// Now send another request for the same chain id but different tx hash.  This should get dropped.
+	req2 := &gossipv1.ObservationRequest{
+		ChainId: 1,
+		TxHash:  []byte{0x96, 0xe3, 0x94, 0xec, 0x5a, 0x00, 0xfc, 0x8b},
+	}
+	ctx.obsvReqC <- req2
+
+	// This is a bit awkward but we need to wait until the goroutine handling the requests has finished
+	// processing the second request.  If we read from the channel too quickly then we might pop out the
+	// first request too early, unblocking the channel.  Unfortunately there's no easy way for us to detect
+	// when the handler is done without adding unnecessary complexity.
+	time.Sleep(50 * time.Millisecond)
+
+	actual, ok := readFromChannel(ctx, ctx.chainObsvReqC[vaa.ChainID(req.ChainId)])
+	assert.True(t, ok)
+	assert.Equal(t, req, actual)
+
+	_, ok = readFromChannel(ctx, ctx.chainObsvReqC[vaa.ChainID(req2.ChainId)])
+	assert.False(t, ok)
+}

node/pkg/processor/cleanup.go

@@ -56,6 +56,7 @@ var (
 
 const (
 	settlementTime = time.Second * 30
+	retryTime      = time.Minute * 5
 )
 
 // handleCleanup handles periodic retransmissions and cleanup of observations
@@ -175,7 +176,7 @@ func (p *Processor) handleCleanup(ctx context.Context) {
 			p.logger.Info("expiring unsubmitted observation after exhausting retries", zap.String("digest", hash), zap.Duration("delta", delta))
 			delete(p.state.signatures, hash)
 			aggregationStateTimeout.Inc()
-		case !s.submitted && delta.Minutes() >= 5:
+		case !s.submitted && delta.Minutes() >= 5 && time.Since(s.lastRetry) >= retryTime:
 			// Poor observation has been unsubmitted for five minutes - clearly, something went wrong.
 			// If we have previously submitted an observation, we can make another attempt to get it over
 			// the finish line by sending a re-observation request to the network and rebroadcasting our
@@ -196,6 +197,7 @@ func (p *Processor) handleCleanup(ctx context.Context) {
 				}
 				p.sendC <- s.ourMsg
 				s.retryCount += 1
+				s.lastRetry = time.Now()
 				aggregationStateRetries.Inc()
 			} else {
 				// For nil state entries, we log the quorum to determine whether the

node/pkg/processor/processor.go

@@ -40,6 +40,8 @@ type (
 	state struct {
 		// First time this digest was seen (possibly even before we observed it ourselves).
 		firstObserved time.Time
+		// The most recent time that a re-observation request was sent to the guardian network.
+		lastRetry time.Time
 		// Copy of our observation.
 		ourObservation Observation
 		// Map of signatures seen by guardian. During guardian set updates, this may contain signatures belonging