pagination for polling, fix tests

eibc/order_client_test.go

@@ -148,22 +148,25 @@ func TestOrderClient(t *testing.T) {
 			pollOrders: []Order{
 				{
 					EibcOrderId: "order1",
-					Price:       "80stake",
+					Price:       "80",
 					Fee:         "12stake",
 					RollappId:   "rollapp1",
 					ProofHeight: "1",
+					BlockHeight: "1",
 				}, {
 					EibcOrderId: "order2",
-					Price:       "202stake",
+					Price:       "202",
 					Fee:         "25stake",
 					RollappId:   "rollapp2",
 					ProofHeight: "2",
+					BlockHeight: "2",
 				}, {
 					EibcOrderId: "order5",
-					Price:       "201stake",
+					Price:       "201",
 					Fee:         "50stake",
 					RollappId:   "rollapp1",
 					ProofHeight: "5",
+					BlockHeight: "5",
 				},
 			},
 			eventOrders: []Order{

eibc/order_poller.go

@@ -23,10 +23,9 @@ type orderPoller struct {
 	indexerClient *http.Client
 	logger        *zap.Logger
 
-	getOrders    func() ([]Order, error)
-	orderTracker *orderTracker
-
-	skippedOrders map[string]struct{}
+	getOrders       func() ([]Order, error)
+	orderTracker    *orderTracker
+	lastBlockHeight uint64
 }
 
 const maxSkippedOrders = 1000
@@ -44,14 +43,13 @@ func newOrderPoller(
 		logger:        logger.With(zap.String("module", "order-poller")),
 		orderTracker:  orderTracker,
 		indexerClient: &http.Client{Timeout: 25 * time.Second},
-		skippedOrders: make(map[string]struct{}),
 	}
 	o.getOrders = o.getDemandOrdersFromIndexer
 	return o
 }
 
 const (
-	ordersQuery = `{"query": "{ibcTransferDetails(filter: {network: {equalTo: \"%s\"} status: {equalTo: EibcPending}}) {nodes { eibcOrderId amount proofHeight price rollappId eibcFee }}}"}`
+	ordersQuery = `{"query": "{ibcTransferDetails(filter: {network: {equalTo: \"%s\"} status: {equalTo: EibcPending}, blockHeight: { greaterThan: \"%s\" }}) {nodes { eibcOrderId amount proofHeight blockHeight price rollappId eibcFee }}}"}`
 )
 
 type Order struct {
@@ -61,6 +59,7 @@ type Order struct {
 	Fee         string `json:"eibcFee"`
 	RollappId   string `json:"rollappId"`
 	ProofHeight string `json:"proofHeight"`
+	BlockHeight string `json:"blockHeight"`
 }
 
 type ordersResponse struct {
@@ -99,16 +98,20 @@ func (p *orderPoller) pollPendingDemandOrders() error {
 
 	demandOrders := make([]Order, 0, len(newDemandOrders))
 	for _, order := range newDemandOrders {
-		if _, ok := p.skippedOrders[order.EibcOrderId]; ok {
+		blockHeight, err := strconv.ParseUint(order.BlockHeight, 10, 64)
+		if err != nil {
+			p.logger.Error("failed to parse block height", zap.Error(err))
 			continue
 		}
+		if blockHeight > p.lastBlockHeight {
+			p.lastBlockHeight = blockHeight
+		}
 		demandOrders = append(demandOrders, order)
 	}
 
 	newOrders := p.convertOrders(demandOrders)
 
 	if len(newOrders) == 0 {
-		p.logger.Debug("no new orders")
 		return nil
 	}
 
@@ -181,9 +184,6 @@ func (p *orderPoller) convertOrders(demandOrders []Order) (orders []*demandOrder
 
 		if err := p.orderTracker.findLPForOrder(newOrder); err != nil {
 			p.logger.Debug("failed to find LP for order", zap.Error(err), zap.String("order_id", newOrder.id))
-			if len(p.skippedOrders) < maxSkippedOrders {
-				p.skippedOrders[newOrder.id] = struct{}{}
-			}
 			continue
 		}
 
@@ -197,9 +197,7 @@ func (p *orderPoller) convertOrders(demandOrders []Order) (orders []*demandOrder
 }
 
 func (p *orderPoller) getDemandOrdersFromIndexer() ([]Order, error) {
-	p.logger.Debug("getting demand orders from indexer")
-
-	queryStr := fmt.Sprintf(ordersQuery, p.chainID)
+	queryStr := fmt.Sprintf(ordersQuery, p.chainID, fmt.Sprint(p.lastBlockHeight))
 	body := strings.NewReader(queryStr)
 
 	resp, err := p.indexerClient.Post(p.indexerURL, "application/json", body)