services/horizon/internal/db2/history: Implement StreamAllOffers using batches

services/horizon/internal/db2/history/offers.go

@@ -2,13 +2,16 @@ package history
 
 import (
 	"context"
+	"math"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/jmoiron/sqlx"
 
 	"github.com/stellar/go/support/errors"
 )
 
+const offersBatchSize = 50000
+
 // QOffers defines offer related queries.
 type QOffers interface {
 	StreamAllOffers(ctx context.Context, callback func(Offer) error) error
@@ -83,28 +86,45 @@ func (q *Q) GetOffers(ctx context.Context, query OffersQuery) ([]Offer, error) {
 
 // StreamAllOffers loads all non deleted offers
 func (q *Q) StreamAllOffers(ctx context.Context, callback func(Offer) error) error {
+	lastID := int64(0)
+	for {
+		nextID, err := q.streamAllOffersBatch(ctx, lastID, offersBatchSize, callback)
+		if err != nil {
+			return err
+		}
+		if lastID == nextID {
+			return nil
+		}
+		lastID = nextID
+	}
+}
+
+func (q *Q) streamAllOffersBatch(ctx context.Context, lastId int64, limit uint64, callback func(Offer) error) (int64, error) {
 	var rows *sqlx.Rows
 	var err error
 
-	if rows, err = q.Query(ctx, selectOffers.Where("deleted = ?", false)); err != nil {
-		return errors.Wrap(err, "could not run all offers select query")
+	rows, err = q.Query(ctx, selectOffers.
+		Where("deleted = ?", false).
+		Where("offer_id > ? ", lastId).
+		OrderBy("offer_id asc").Limit(limit))
+	if err != nil {
+		return 0, errors.Wrap(err, "could not run all offers select query")
 	}
 
 	defer rows.Close()
-
 	for rows.Next() {
 		offer := Offer{}
 		if err = rows.StructScan(&offer); err != nil {
-			return errors.Wrap(err, "could not scan row into offer struct")
+			return 0, errors.Wrap(err, "could not scan row into offer struct")
 		}
 
 		if err = callback(offer); err != nil {
-			return err
+			return 0, err
 		}
+		lastId = offer.OfferID
 	}
 
-	return rows.Err()
-
+	return lastId, rows.Err()
 }
 
 // GetUpdatedOffers returns all offers created, updated, or deleted after the given ledger sequence.

services/horizon/internal/ingest/orderbook.go

@@ -136,7 +136,6 @@ func (o *OrderBookStream) update(ctx context.Context, status ingestionStatus) (b
 			o.graph.AddOffers(offerToXDR(offer))
 			return nil
 		})
-
 		if err != nil {
 			return true, errors.Wrap(err, "Error loading offers into orderbook")
 		}