Merge pull request #613 from ava-labs/timeout-backoff-ops

Add timeout to remote ops

relayer/application_relayer.go

@@ -28,7 +28,8 @@ import (
 )
 
 const (
-	retryTimeout = 10 * time.Second
+	retryTimeout   = 10 * time.Second
+	warpAPITimeout = utils.DefaultRPCTimeout + peers.DefaultAppRequestTimeout
 )
 
 // Errors
@@ -270,9 +271,11 @@ func (r *ApplicationRelayer) createSignedMessage(
 		signedWarpMessageBytes hexutil.Bytes
 		err                    error
 	)
+	cctx, cancel := context.WithTimeout(context.Background(), warpAPITimeout)
+	defer cancel()
 	operation := func() error {
 		return r.sourceWarpSignatureClient.CallContext(
-			context.Background(),
+			cctx,
 			&signedWarpMessageBytes,
 			"warp_getMessageAggregateSignature",
 			unsignedMessage.ID(),

utils/backoff.go

@@ -8,7 +8,8 @@ import (
 )
 
 // WithRetriesTimeout uses an exponential backoff to run the operation until it
-// succeeds or timeout limit has been reached.
+// succeeds or timeout limit has been reached. It is the caller's responsibility
+// to ensure {operation} returns. It is safe for {operation} to take longer than {timeout}.
 func WithRetriesTimeout(
 	logger logging.Logger,
 	operation backoff.Operation,

vms/evm/subscriber.go

@@ -125,10 +125,14 @@ func (s *subscriber) processBlockRange(
 }
 
 func (s *subscriber) getHeaderByNumberRetryable(headerNumber *big.Int) (*types.Header, error) {
-	var err error
-	var header *types.Header
+	var (
+		err    error
+		header *types.Header
+	)
+	cctx, cancel := context.WithTimeout(context.Background(), utils.DefaultRPCTimeout)
+	defer cancel()
 	operation := func() (err error) {
-		header, err = s.rpcClient.HeaderByNumber(context.Background(), headerNumber)
+		header, err = s.rpcClient.HeaderByNumber(cctx, headerNumber)
 		return err
 	}
 	err = utils.WithRetriesTimeout(s.logger, operation, utils.DefaultRPCTimeout)
@@ -161,8 +165,10 @@ func (s *subscriber) Subscribe(retryTimeout time.Duration) error {
 // subscribe until it succeeds or reached timeout.
 func (s *subscriber) subscribe(retryTimeout time.Duration) error {
 	var sub interfaces.Subscription
+	cctx, cancel := context.WithTimeout(context.Background(), utils.DefaultRPCTimeout)
+	defer cancel()
 	operation := func() (err error) {
-		sub, err = s.wsClient.SubscribeNewHead(context.Background(), s.headers)
+		sub, err = s.wsClient.SubscribeNewHead(cctx, s.headers)
 		return err
 	}
 	err := utils.WithRetriesTimeout(s.logger, operation, retryTimeout)