fix: use context timeout per paginated query (#1395)

* fix: use context timeout per paginated query

Previously we were using one 60s timeout for the entire paginated query request for channels. This change moves the context timeout to be on a per-request basis and uses a 10s timeout per request.

* refactor: break out anonymous function into a named method to use in both channel query methods

* refactor: avoid redundant code and use QueryChannelsPaginated in QueryChannels

relayer/chains/cosmos/cosmos_chain_processor.go

@@ -307,12 +307,14 @@ func (ccp *CosmosChainProcessor) initializeConnectionState(ctx context.Context)
 
 // initializeChannelState will bootstrap the channelStateCache with the open channel state.
 func (ccp *CosmosChainProcessor) initializeChannelState(ctx context.Context) error {
-	ctx, cancel := context.WithTimeout(ctx, queryStateTimeout)
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
+
 	channels, err := ccp.chainProvider.QueryChannels(ctx)
 	if err != nil {
 		return fmt.Errorf("error querying channels: %w", err)
 	}
+
 	for _, ch := range channels {
 		if len(ch.ConnectionHops) != 1 {
 			ccp.log.Error("Found channel using multiple connection hops. Not currently supported, ignoring.",
@@ -322,15 +324,18 @@ func (ccp *CosmosChainProcessor) initializeChannelState(ctx context.Context) err
 			)
 			continue
 		}
+
 		ccp.channelConnections[ch.ChannelId] = ch.ConnectionHops[0]
 		k := processor.ChannelKey{
 			ChannelID:             ch.ChannelId,
 			PortID:                ch.PortId,
 			CounterpartyChannelID: ch.Counterparty.ChannelId,
 			CounterpartyPortID:    ch.Counterparty.PortId,
 		}
+
 		ccp.channelStateCache.SetOpen(k, ch.State == chantypes.OPEN, ch.Ordering)
 	}
+
 	return nil
 }
 

relayer/chains/cosmos/query.go

@@ -882,36 +882,38 @@ func (cc *CosmosProvider) QueryConnectionChannels(ctx context.Context, height in
 	return channels, nil
 }
 
-// QueryChannels returns all the channels that are registered on a chain
+// QueryChannels returns all the channels that are registered on a chain.
 func (cc *CosmosProvider) QueryChannels(ctx context.Context) ([]*chantypes.IdentifiedChannel, error) {
-	qc := chantypes.NewQueryClient(cc)
 	p := DefaultPageRequest()
 	chans := []*chantypes.IdentifiedChannel{}
 
 	for {
-		res, err := qc.Channels(ctx, &chantypes.QueryChannelsRequest{
-			Pagination: p,
-		})
+		res, next, err := cc.QueryChannelsPaginated(ctx, p)
 		if err != nil {
 			return nil, err
 		}
 
-		chans = append(chans, res.Channels...)
-		next := res.GetPagination().GetNextKey()
+		chans = append(chans, res...)
 		if len(next) == 0 {
 			break
 		}
 
 		time.Sleep(PaginationDelay)
 		p.Key = next
 	}
+
 	return chans, nil
 }
 
-// QueryChannels returns all the channels that are registered on a chain
-func (cc *CosmosProvider) QueryChannelsPaginated(ctx context.Context, pageRequest *querytypes.PageRequest) ([]*chantypes.IdentifiedChannel, []byte, error) {
+// QueryChannelsPaginated returns all the channels for a particular paginated request that are registered on a chain.
+func (cc *CosmosProvider) QueryChannelsPaginated(
+	ctx context.Context,
+	pageRequest *querytypes.PageRequest,
+) ([]*chantypes.IdentifiedChannel, []byte, error) {
 	qc := chantypes.NewQueryClient(cc)
-	chans := []*chantypes.IdentifiedChannel{}
+
+	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
+	defer cancel()
 
 	res, err := qc.Channels(ctx, &chantypes.QueryChannelsRequest{
 		Pagination: pageRequest,
@@ -920,10 +922,9 @@ func (cc *CosmosProvider) QueryChannelsPaginated(ctx context.Context, pageReques
 		return nil, nil, err
 	}
 
-	chans = append(chans, res.Channels...)
 	next := res.GetPagination().GetNextKey()
 
-	return chans, next, nil
+	return res.Channels, next, nil
 }
 
 // QueryPacketCommitments returns an array of packet commitments