BE-586 | Requested changes #1

ingest/usecase/plugins/orderbook/claimbot/export_test.go

@@ -20,8 +20,8 @@ import (
 	txtypes "github.com/cosmos/cosmos-sdk/types/tx"
 )
 
-// Order is order alias data structure for testing purposes.
-type Order = order
+// ProcessedOrderbook is order alias data structure for testing purposes.
+type ProcessedOrderbook = processedOrderbook
 
 // ProcessOrderbooksAndGetClaimableOrders is test wrapper for processOrderbooksAndGetClaimableOrders.
 // This function is exported for testing purposes.
@@ -33,7 +33,7 @@ func ProcessOrderbooksAndGetClaimableOrders(
 	orderBookClient orderbookgrpcclientdomain.OrderBookClient,
 	orderbookusecase mvc.OrderBookUsecase,
 	logger log.Logger,
-) []Order {
+) ([]ProcessedOrderbook, error) {
 	return processOrderbooksAndGetClaimableOrders(ctx, fillThreshold, orderbooks, orderbookRepository, orderBookClient, orderbookusecase, logger)
 }
 
@@ -61,6 +61,6 @@ func PrepareBatchClaimMsg(claims orderbookdomain.Orders) ([]byte, error) {
 
 // GetOrderbooks is a test wrapper for getOrderbooks.
 // This function is exported for testing purposes.
-func GetOrderbooks(poolsUsecase mvc.PoolsUsecase, blockHeight uint64, metadata domain.BlockPoolMetadata) ([]domain.CanonicalOrderBooksResult, error) {
-	return getOrderbooks(poolsUsecase, blockHeight, metadata)
+func GetOrderbooks(poolsUsecase mvc.PoolsUsecase, metadata domain.BlockPoolMetadata) ([]domain.CanonicalOrderBooksResult, error) {
+	return getOrderbooks(poolsUsecase, metadata)
 }

ingest/usecase/plugins/orderbook/claimbot/order.go

@@ -14,13 +14,14 @@ import (
 	"go.uber.org/zap"
 )
 
-type order struct {
+type processedOrderbook struct {
 	Orderbook domain.CanonicalOrderBooksResult
 	Orders    orderbookdomain.Orders
 	Err       error
 }
 
 // processOrderbooksAndGetClaimableOrders processes a list of orderbooks and returns claimable orders for each.
+// Under the hood processing of each orderbook in done concurrently to speed up the process.
 func processOrderbooksAndGetClaimableOrders(
 	ctx context.Context,
 	fillThreshold osmomath.Dec,
@@ -29,13 +30,27 @@ func processOrderbooksAndGetClaimableOrders(
 	orderBookClient orderbookgrpcclientdomain.OrderBookClient,
 	orderbookusecase mvc.OrderBookUsecase,
 	logger log.Logger,
-) []order {
-	var result []order
+) ([]processedOrderbook, error) {
+	ch := make(chan processedOrderbook, len(orderbooks))
+
 	for _, orderbook := range orderbooks {
-		processedOrder := processOrderbook(ctx, fillThreshold, orderbook, orderbookRepository, orderBookClient, orderbookusecase, logger)
-		result = append(result, processedOrder)
+		go func(orderbook domain.CanonicalOrderBooksResult) {
+			o := processOrderbook(ctx, fillThreshold, orderbook, orderbookRepository, orderBookClient, orderbookusecase, logger)
+			ch <- o
+		}(orderbook)
 	}
-	return result
+
+	var results []processedOrderbook
+	for i := 0; i < len(orderbooks); i++ {
+		select {
+		case result := <-ch:
+			results = append(results, result)
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+	return results, nil
 }
 
 // processOrderbook processes a single orderbook and returns an order struct containing the processed orderbook and its claimable orders.
@@ -47,15 +62,15 @@ func processOrderbook(
 	orderBookClient orderbookgrpcclientdomain.OrderBookClient,
 	orderbookusecase mvc.OrderBookUsecase,
 	logger log.Logger,
-) order {
+) processedOrderbook {
 	claimable, err := getClaimableOrdersForOrderbook(ctx, fillThreshold, orderbook, orderbookRepository, orderBookClient, orderbookusecase, logger)
 	if err != nil {
-		return order{
+		return processedOrderbook{
 			Orderbook: orderbook,
 			Err:       err,
 		}
 	}
-	return order{
+	return processedOrderbook{
 		Orderbook: orderbook,
 		Orders:    claimable,
 	}
@@ -75,14 +90,19 @@ func getClaimableOrdersForOrderbook(
 ) (orderbookdomain.Orders, error) {
 	ticks, ok := orderbookRepository.GetAllTicks(orderbook.PoolID)
 	if !ok {
-		return nil, fmt.Errorf("no ticks for orderbook")
+		return nil, fmt.Errorf("no ticks found for orderbook %s with pool %d", orderbook.ContractAddress, orderbook.PoolID)
 	}
 
 	var claimable orderbookdomain.Orders
-	for _, t := range ticks {
-		tickClaimable, err := getClaimableOrdersForTick(ctx, fillThreshold, orderbook, t, orderBookClient, orderbookusecase, logger)
+	for _, tick := range ticks {
+		tickClaimable, err := getClaimableOrdersForTick(ctx, fillThreshold, orderbook, tick, orderBookClient, orderbookusecase, logger)
 		if err != nil {
-			logger.Error("error processing tick", zap.String("orderbook", orderbook.ContractAddress), zap.Int64("tick", t.Tick.TickId), zap.Error(err))
+			logger.Error(
+				"error processing tick",
+				zap.String("orderbook", orderbook.ContractAddress),
+				zap.Int64("tick", tick.Tick.TickId),
+				zap.Error(err),
+			)
 			continue
 		}
 		claimable = append(claimable, tickClaimable...)
@@ -104,7 +124,7 @@ func getClaimableOrdersForTick(
 ) (orderbookdomain.Orders, error) {
 	orders, err := orderBookClient.GetOrdersByTick(ctx, orderbook.ContractAddress, tick.Tick.TickId)
 	if err != nil {
-		return nil, fmt.Errorf("unable to fetch orderbook orders by tick ID: %w", err)
+		return nil, err
 	}
 
 	if len(orders) == 0 {
@@ -138,10 +158,17 @@ func getClaimableOrders(
 // isTickFullyFilled checks if a tick is fully filled by comparing its cumulative total value
 // to its effective total amount swapped.
 func isTickFullyFilled(tickValues orderbookdomain.TickValues) bool {
-	if len(tickValues.CumulativeTotalValue) == 0 || len(tickValues.EffectiveTotalAmountSwapped) == 0 {
-		return false // empty values, thus not fully filled
+	cumulativeTotalValue, err := osmomath.NewDecFromStr(tickValues.CumulativeTotalValue)
+	if err != nil {
+		return false // if the cumulative total value is invalid, we assume the tick is not fully filled
 	}
-	return tickValues.CumulativeTotalValue == tickValues.EffectiveTotalAmountSwapped
+
+	effectiveTotalAmountSwapped, err := osmomath.NewDecFromStr(tickValues.EffectiveTotalAmountSwapped)
+	if err != nil {
+		return false // if the effective total amount swapped is invalid, we assume the tick is not fully filled
+	}
+
+	return cumulativeTotalValue.Equal(effectiveTotalAmountSwapped)
 }
 
 // filterClaimableOrders processes a list of orders and returns only those that are considered claimable.

ingest/usecase/plugins/orderbook/claimbot/order_test.go

@@ -74,7 +74,7 @@ func TestProcessOrderbooksAndGetClaimableOrders(t *testing.T) {
 		fillThreshold  osmomath.Dec
 		orderbooks     []domain.CanonicalOrderBooksResult
 		mockSetup      func(*mocks.OrderbookRepositoryMock, *mocks.OrderbookGRPCClientMock, *mocks.OrderbookUsecaseMock)
-		expectedOrders []claimbot.Order
+		expectedOrders []claimbot.ProcessedOrderbook
 	}{
 		{
 			name:          "No orderbooks",
@@ -103,7 +103,7 @@ func TestProcessOrderbooksAndGetClaimableOrders(t *testing.T) {
 				// Not claimable order, below threshold
 				usecase.WithCreateFormattedLimitOrder(newLimitOrder(osmomath.NewDecWithPrec(90, 2)), nil)
 			},
-			expectedOrders: []claimbot.Order{
+			expectedOrders: []claimbot.ProcessedOrderbook{
 				{
 					Orderbook: newCanonicalOrderBooksResult(10, "contract1"), // orderbook with
 					Orders:    nil,                                           // no claimable orders
@@ -125,7 +125,7 @@ func TestProcessOrderbooksAndGetClaimableOrders(t *testing.T) {
 
 				usecase.WithCreateFormattedLimitOrder(newLimitOrder(osmomath.NewDecWithPrec(90, 2)), nil)
 			},
-			expectedOrders: []claimbot.Order{
+			expectedOrders: []claimbot.ProcessedOrderbook{
 				{
 					Orderbook: newCanonicalOrderBooksResult(38, "contract8"),
 					Orders:    orderbookdomain.Orders{newOrder("bid")},
@@ -143,15 +143,19 @@ func TestProcessOrderbooksAndGetClaimableOrders(t *testing.T) {
 
 				client.WithGetOrdersByTickCb(orderbookdomain.Orders{
 					newOrder("ask"),
+					newOrder("bid"),
 				}, nil)
 
 				// Claimable order, above threshold
 				usecase.WithCreateFormattedLimitOrder(newLimitOrder(osmomath.NewDecWithPrec(96, 2)), nil)
 			},
-			expectedOrders: []claimbot.Order{
+			expectedOrders: []claimbot.ProcessedOrderbook{
 				{
 					Orderbook: newCanonicalOrderBooksResult(64, "contract58"),
-					Orders:    orderbookdomain.Orders{newOrder("ask")},
+					Orders: orderbookdomain.Orders{
+						newOrder("ask"),
+						newOrder("bid"),
+					},
 				},
 			},
 		},
@@ -167,7 +171,8 @@ func TestProcessOrderbooksAndGetClaimableOrders(t *testing.T) {
 
 			tt.mockSetup(&repository, &client, &usecase)
 
-			result := claimbot.ProcessOrderbooksAndGetClaimableOrders(ctx, tt.fillThreshold, tt.orderbooks, &repository, &client, &usecase, &logger)
+			result, err := claimbot.ProcessOrderbooksAndGetClaimableOrders(ctx, tt.fillThreshold, tt.orderbooks, &repository, &client, &usecase, &logger)
+			assert.NoError(t, err)
 
 			assert.Equal(t, tt.expectedOrders, result)
 		})

ingest/usecase/plugins/orderbook/claimbot/orderbook.go

@@ -1,17 +1,15 @@
 package claimbot
 
 import (
-	"fmt"
-
 	"github.com/osmosis-labs/sqs/domain"
 	"github.com/osmosis-labs/sqs/domain/mvc"
 )
 
 // getOrderbooks returns canonical orderbooks that are within the metadata.
-func getOrderbooks(poolsUsecase mvc.PoolsUsecase, blockHeight uint64, metadata domain.BlockPoolMetadata) ([]domain.CanonicalOrderBooksResult, error) {
+func getOrderbooks(poolsUsecase mvc.PoolsUsecase, metadata domain.BlockPoolMetadata) ([]domain.CanonicalOrderBooksResult, error) {
 	orderbooks, err := poolsUsecase.GetAllCanonicalOrderbookPoolIDs()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get all canonical orderbook pool IDs ( block height %d ) : %w", blockHeight, err)
+		return nil, err
 	}
 
 	var result []domain.CanonicalOrderBooksResult

ingest/usecase/plugins/orderbook/claimbot/orderbook_test.go

@@ -12,16 +12,14 @@ import (
 
 func TestGetOrderbooks(t *testing.T) {
 	tests := []struct {
-		name        string
-		blockHeight uint64
-		metadata    domain.BlockPoolMetadata
-		setupMocks  func(*mocks.PoolsUsecaseMock)
-		want        []domain.CanonicalOrderBooksResult
-		err         bool
+		name       string
+		metadata   domain.BlockPoolMetadata
+		setupMocks func(*mocks.PoolsUsecaseMock)
+		want       []domain.CanonicalOrderBooksResult
+		err        bool
 	}{
 		{
-			name:        "Metadata contains all canonical orderbooks but one",
-			blockHeight: 1000,
+			name: "Metadata contains all canonical orderbooks but one",
 			metadata: domain.BlockPoolMetadata{
 				PoolIDs: map[uint64]struct{}{1: {}, 2: {}, 3: {}},
 			},
@@ -36,8 +34,7 @@ func TestGetOrderbooks(t *testing.T) {
 			err: false,
 		},
 		{
-			name:        "Metadata contains only canonical orderbooks",
-			blockHeight: 1893,
+			name: "Metadata contains only canonical orderbooks",
 			metadata: domain.BlockPoolMetadata{
 				PoolIDs: map[uint64]struct{}{1: {}, 2: {}, 3: {}},
 			},
@@ -52,9 +49,8 @@ func TestGetOrderbooks(t *testing.T) {
 			err: false,
 		},
 		{
-			name:        "Error getting all canonical orderbook pool IDs",
-			blockHeight: 2000,
-			metadata:    domain.BlockPoolMetadata{},
+			name:     "Error getting all canonical orderbook pool IDs",
+			metadata: domain.BlockPoolMetadata{},
 			setupMocks: func(poolsUsecase *mocks.PoolsUsecaseMock) {
 				poolsUsecase.WithGetAllCanonicalOrderbookPoolIDs(nil, assert.AnError)
 			},
@@ -69,7 +65,7 @@ func TestGetOrderbooks(t *testing.T) {
 
 			tt.setupMocks(&poolsUsecase)
 
-			got, err := claimbot.GetOrderbooks(&poolsUsecase, tt.blockHeight, tt.metadata)
+			got, err := claimbot.GetOrderbooks(&poolsUsecase, tt.metadata)
 			if tt.err {
 				assert.Error(t, err)
 				return

ingest/usecase/plugins/orderbook/claimbot/plugin.go

@@ -79,14 +79,20 @@ func (o *claimbot) ProcessEndBlock(ctx context.Context, blockHeight uint64, meta
 		return nil
 	}
 	defer o.atomicBool.Store(false)
+	defer o.config.Logger.Info("processed end block", zap.Uint64("block_height", blockHeight))
 
-	orderbooks, err := getOrderbooks(o.config.PoolsUseCase, blockHeight, metadata)
+	orderbooks, err := getOrderbooks(o.config.PoolsUseCase, metadata)
 	if err != nil {
+		o.config.Logger.Warn(
+			"failed to get canonical orderbook pools for block",
+			zap.Uint64("block_height", blockHeight),
+			zap.Error(err),
+		)
 		return err
 	}
 
 	// retrieve claimable orders for the orderbooks
-	orders := processOrderbooksAndGetClaimableOrders(
+	orders, err := processOrderbooksAndGetClaimableOrders(
 		ctx,
 		fillThreshold,
 		orderbooks,
@@ -96,6 +102,14 @@ func (o *claimbot) ProcessEndBlock(ctx context.Context, blockHeight uint64, meta
 		o.config.Logger,
 	)
 
+	if err != nil {
+		o.config.Logger.Warn(
+			"failed to process block orderbooks",
+			zap.Error(err),
+		)
+		return err
+	}
+
 	for _, orderbook := range orders {
 		if orderbook.Err != nil {
 			o.config.Logger.Warn(
@@ -115,15 +129,13 @@ func (o *claimbot) ProcessEndBlock(ctx context.Context, blockHeight uint64, meta
 		}
 	}
 
-	o.config.Logger.Info("processed end block", zap.Uint64("block_height", blockHeight))
-
 	return nil
 }
 
 // processOrderbookOrders processes a batch of claimable orders.
 func (o *claimbot) processOrderbookOrders(ctx context.Context, orderbook domain.CanonicalOrderBooksResult, orders orderbookdomain.Orders) error {
 	if len(orders) == 0 {
-		return fmt.Errorf("no claimable orders found for orderbook %s, nothing to process", orderbook.ContractAddress)
+		return nil
 	}
 
 	for _, chunk := range slices.Split(orders, maxBatchOfClaimableOrders) {

ingest/usecase/plugins/orderbook/claimbot/tx.go

@@ -62,21 +62,26 @@ func sendBatchClaimTx(
 	return sqstx.SendTx(ctx, txServiceClient, txBytes)
 }
 
+// batchClaim represents batch claim orders message.
+type batchClaim struct {
+	batchClaimOrders `json:"batch_claim"`
+}
+
+// batchClaimOrders represents the orders in the batch claim message.
+// Each order is represented by a pair of tick ID and order ID.
+type batchClaimOrders struct {
+	Orders [][]int64 `json:"orders"`
+}
+
 // prepareBatchClaimMsg creates a JSON-encoded batch claim message from the provided orders.
 func prepareBatchClaimMsg(claims orderbookdomain.Orders) ([]byte, error) {
 	orders := make([][]int64, len(claims))
 	for i, claim := range claims {
 		orders[i] = []int64{claim.TickId, claim.OrderId}
 	}
 
-	batchClaim := struct {
-		BatchClaim struct {
-			Orders [][]int64 `json:"orders"`
-		} `json:"batch_claim"`
-	}{
-		BatchClaim: struct {
-			Orders [][]int64 `json:"orders"`
-		}{
+	batchClaim := batchClaim{
+		batchClaimOrders: batchClaimOrders{
 			Orders: orders,
 		},
 	}

ingest/usecase/plugins/orderbook/fillbot/create_copy_config.sh

@@ -4,6 +4,11 @@
 ORIGINAL_APP_TOML_NAME="$HOME/.osmosisd/config/app.toml"   # Replace with the actual file path
 BACKUP_APP_TOML_NAME="$HOME/.osmosisd/config/app-backup.toml"
 
+if [ ! -f "$ORIGINAL_APP_TOML_NAME" ]; then
+	echo "Error: Source file $ORIGINAL_APP_TOML_NAME does not exist."
+	exit 1
+fi
+
 if [ -f $BACKUP_APP_TOML_NAME ]; then
 	echo "Backup file $BACKUP_APP_TOML_NAME already exist, no modifications will be made."
 	exit 0