[CT-856] Add order replacement to fix vault causing orderbook flickering

indexer/services/vulcan/__tests__/handlers/order-replace-handler.test.ts

@@ -96,7 +96,6 @@ describe('order-replace-handler', () => {
       ...replacementOrderGoodTilBlockTime,
       subticks: replacementOrderGoodTilBlockTime.subticks.mul(2),
     };
-
     const replacedOrder: RedisOrder = redisPackage.convertToRedisOrder(
       replacementOrder,
       testConstants.defaultPerpetualMarket,
@@ -185,6 +184,7 @@ describe('order-replace-handler', () => {
       jest.spyOn(CanceledOrdersCache, 'removeOrderFromCaches');
       jest.spyOn(stats, 'increment');
       jest.spyOn(redisPackage, 'placeOrder');
+      jest.spyOn(redisPackage, 'removeOrder');
       jest.spyOn(logger, 'error');
       jest.spyOn(logger, 'info');
     });
@@ -559,6 +559,22 @@ describe('order-replace-handler', () => {
       },
     );
 
+    it('replaces order successfully where old order does not exist', async () => {
+      synchronizeWrapBackgroundTask(wrapBackgroundTask);
+      const producerSendSpy: jest.SpyInstance = jest.spyOn(producer, 'send').mockReturnThis();
+      await onMessage(replacementMessage);
+
+      expect(redisPackage.removeOrder).toHaveBeenCalled();
+      expect(redisPackage.placeOrder).toHaveBeenCalled();
+      expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({
+        at: 'OrderReplaceHandler#handle',
+        message: 'Old order not found in cache',
+        oldOrderId: redisTestConstants.defaultOrderId,
+      }));
+      expect(OrderbookLevelsCache.updatePriceLevel).not.toHaveBeenCalled();
+      expectWebsocketMessagesNotSent(producerSendSpy);
+    });
+
     it.each([
       [
         'missing order',
@@ -675,9 +691,11 @@ async function checkOrderReplace(
   placedSubaccountId: string,
   expectedOrder: RedisOrder,
 ): Promise<void> {
+  expect(redisPackage.removeOrder).toHaveBeenCalled();
   const oldRedisOrder: RedisOrder | null = await OrdersCache.getOrder(oldOrderId, client);
   expect(oldRedisOrder).toBeNull();
 
+  expect(redisPackage.placeOrder).toHaveBeenCalled();
   const newRedisOrder: RedisOrder | null = await OrdersCache.getOrder(placedOrderId, client);
   const orderIdsForSubaccount: string[] = await SubaccountOrderIdsCache.getOrderIdsForSubaccount(
     placedSubaccountId,

indexer/services/vulcan/src/handlers/order-replace-handler.ts

@@ -71,6 +71,17 @@ export class OrderReplaceHandler extends Handler {
     /* Remove old order */
     const removeOrderResult: RemoveOrderResult = await this.removeOldOrder(oldOrderId);
 
+    /* We don't want to fail if old order is not found (new order should still be placed),
+    so log and track metric */
+    if (!removeOrderResult.removed) {
+      logger.info({
+        at: 'OrderReplaceHandler#handle',
+        message: 'Old order not found in cache',
+        oldOrderId,
+      });
+      stats.increment(`${config.SERVICE_NAME}.replace_order_handler.old_order_not_found_in_cache`, 1);
+    }
+
     /* Place new order */
     const order: IndexerOrder = orderReplace.order!;
     const placementStatus: OrderPlaceV1_OrderPlacementStatus = orderReplace.placementStatus;

protocol/app/app.go

@@ -1097,6 +1097,7 @@ func New(
 		app.PricesKeeper,
 		app.SendingKeeper,
 		app.SubaccountsKeeper,
+		app.IndexerEventManager,
 		[]string{
 			lib.GovModuleAddress.String(),
 			delaymsgmoduletypes.ModuleAddress.String(),

protocol/indexer/events/stateful_order.go

@@ -20,6 +20,23 @@ func NewLongTermOrderPlacementEvent(
 	}
 }
 
+func NewLongTermOrderReplacementEvent(
+	oldOrderId clobtypes.OrderId,
+	order clobtypes.Order,
+) *StatefulOrderEventV1 {
+	oldIndexerOrderId := v1.OrderIdToIndexerOrderId(oldOrderId)
+	indexerOrder := v1.OrderToIndexerOrder(order)
+	orderReplace := StatefulOrderEventV1_LongTermOrderReplacementV1{
+		OldOrderId: &oldIndexerOrderId,
+		Order:      &indexerOrder,
+	}
+	return &StatefulOrderEventV1{
+		Event: &StatefulOrderEventV1_OrderReplacement{
+			OrderReplacement: &orderReplace,
+		},
+	}
+}
+
 func NewStatefulOrderRemovalEvent(
 	removedOrderId clobtypes.OrderId,
 	reason sharedtypes.OrderRemovalReason,

protocol/testutil/keeper/vault.go

@@ -58,6 +58,7 @@ func createVaultKeeper(
 		&mocks.PricesKeeper{},
 		&mocks.SendingKeeper{},
 		&mocks.SubaccountsKeeper{},
+		&mocks.IndexerEventManager{},
 		[]string{
 			lib.GovModuleAddress.String(),
 			delaymsgtypes.ModuleAddress.String(),

protocol/x/clob/keeper/msg_server_cancel_orders.go

@@ -23,7 +23,7 @@ func (k msgServer) CancelOrder(
 ) (resp *types.MsgCancelOrderResponse, err error) {
 	ctx := lib.UnwrapSDKContext(goCtx, types.ModuleName)
 
-	if err := k.Keeper.HandleMsgCancelOrder(ctx, msg); err != nil {
+	if err := k.Keeper.HandleMsgCancelOrder(ctx, msg, false); err != nil {
 		return nil, err
 	}
 
@@ -37,6 +37,7 @@ func (k msgServer) CancelOrder(
 func (k Keeper) HandleMsgCancelOrder(
 	ctx sdk.Context,
 	msg *types.MsgCancelOrder,
+	isInternalOrder bool,
 ) (err error) {
 	lib.AssertDeliverTxMode(ctx)
 
@@ -110,17 +111,19 @@ func (k Keeper) HandleMsgCancelOrder(
 	k.MustSetProcessProposerMatchesEvents(ctx, processProposerMatchesEvents)
 
 	// 4. Add the relevant on-chain Indexer event for the cancellation.
-	k.GetIndexerEventManager().AddTxnEvent(
-		ctx,
-		indexerevents.SubtypeStatefulOrder,
-		indexerevents.StatefulOrderEventVersion,
-		indexer_manager.GetBytes(
-			indexerevents.NewStatefulOrderRemovalEvent(
-				msg.OrderId,
-				indexershared.OrderRemovalReason_ORDER_REMOVAL_REASON_USER_CANCELED,
+	if !isInternalOrder { // vault order indexer event logic is handled elsewhere
+		k.GetIndexerEventManager().AddTxnEvent(
+			ctx,
+			indexerevents.SubtypeStatefulOrder,
+			indexerevents.StatefulOrderEventVersion,
+			indexer_manager.GetBytes(
+				indexerevents.NewStatefulOrderRemovalEvent(
+					msg.OrderId,
+					indexershared.OrderRemovalReason_ORDER_REMOVAL_REASON_USER_CANCELED,
+				),
 			),
-		),
-	)
+		)
+	}
 
 	return nil
 }

protocol/x/clob/keeper/msg_server_place_order.go

@@ -95,6 +95,7 @@ func (k Keeper) HandleMsgPlaceOrder(
 			order,
 		)
 	}
+
 	removedOrderIds := lib.UniqueSliceToSet(processProposerMatchesEvents.RemovedStatefulOrderIds)
 	if _, found := removedOrderIds[order.GetOrderId()]; found {
 		return errorsmod.Wrapf(
@@ -114,31 +115,35 @@ func (k Keeper) HandleMsgPlaceOrder(
 
 	// 4. Emit the new order placement indexer event.
 	if order.IsConditionalOrder() {
-		k.GetIndexerEventManager().AddTxnEvent(
-			ctx,
-			indexerevents.SubtypeStatefulOrder,
-			indexerevents.StatefulOrderEventVersion,
-			indexer_manager.GetBytes(
-				indexerevents.NewConditionalOrderPlacementEvent(
-					order,
+		if !isInternalOrder { // vault order indexer event logic is handled elsewhere
+			k.GetIndexerEventManager().AddTxnEvent(
+				ctx,
+				indexerevents.SubtypeStatefulOrder,
+				indexerevents.StatefulOrderEventVersion,
+				indexer_manager.GetBytes(
+					indexerevents.NewConditionalOrderPlacementEvent(
+						order,
+					),
 				),
-			),
-		)
+			)
+		}
 		processProposerMatchesEvents.PlacedConditionalOrderIds = append(
 			processProposerMatchesEvents.PlacedConditionalOrderIds,
 			order.OrderId,
 		)
 	} else {
-		k.GetIndexerEventManager().AddTxnEvent(
-			ctx,
-			indexerevents.SubtypeStatefulOrder,
-			indexerevents.StatefulOrderEventVersion,
-			indexer_manager.GetBytes(
-				indexerevents.NewLongTermOrderPlacementEvent(
-					order,
+		if !isInternalOrder { // vault order indexer event logic is handled elsewhere
+			k.GetIndexerEventManager().AddTxnEvent(
+				ctx,
+				indexerevents.SubtypeStatefulOrder,
+				indexerevents.StatefulOrderEventVersion,
+				indexer_manager.GetBytes(
+					indexerevents.NewLongTermOrderPlacementEvent(
+						order,
+					),
 				),
-			),
-		)
+			)
+		}
 		processProposerMatchesEvents.PlacedLongTermOrderIds = append(
 			processProposerMatchesEvents.PlacedLongTermOrderIds,
 			order.OrderId,

protocol/x/clob/types/clob_keeper.go

@@ -46,6 +46,7 @@ type ClobKeeper interface {
 	HandleMsgCancelOrder(
 		ctx sdk.Context,
 		msg *MsgCancelOrder,
+		isInternalOrder bool,
 	) (err error)
 	HandleMsgPlaceOrder(
 		ctx sdk.Context,

protocol/x/vault/keeper/keeper.go

@@ -7,20 +7,22 @@ import (
 	storetypes "cosmossdk.io/store/types"
 	"github.com/cosmos/cosmos-sdk/codec"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/dydxprotocol/v4-chain/protocol/indexer/indexer_manager"
 	"github.com/dydxprotocol/v4-chain/protocol/lib"
 	"github.com/dydxprotocol/v4-chain/protocol/x/vault/types"
 )
 
 type (
 	Keeper struct {
-		cdc               codec.BinaryCodec
-		storeKey          storetypes.StoreKey
-		clobKeeper        types.ClobKeeper
-		perpetualsKeeper  types.PerpetualsKeeper
-		pricesKeeper      types.PricesKeeper
-		sendingKeeper     types.SendingKeeper
-		subaccountsKeeper types.SubaccountsKeeper
-		authorities       map[string]struct{}
+		cdc                 codec.BinaryCodec
+		storeKey            storetypes.StoreKey
+		clobKeeper          types.ClobKeeper
+		perpetualsKeeper    types.PerpetualsKeeper
+		pricesKeeper        types.PricesKeeper
+		sendingKeeper       types.SendingKeeper
+		subaccountsKeeper   types.SubaccountsKeeper
+		indexerEventManager indexer_manager.IndexerEventManager
+		authorities         map[string]struct{}
 	}
 )
 
@@ -32,20 +34,26 @@ func NewKeeper(
 	pricesKeeper types.PricesKeeper,
 	sendingKeeper types.SendingKeeper,
 	subaccountsKeeper types.SubaccountsKeeper,
+	indexerEventManager indexer_manager.IndexerEventManager,
 	authorities []string,
 ) *Keeper {
 	return &Keeper{
-		cdc:               cdc,
-		storeKey:          storeKey,
-		clobKeeper:        clobKeeper,
-		perpetualsKeeper:  perpetualsKeeper,
-		pricesKeeper:      pricesKeeper,
-		sendingKeeper:     sendingKeeper,
-		subaccountsKeeper: subaccountsKeeper,
-		authorities:       lib.UniqueSliceToSet(authorities),
+		cdc:                 cdc,
+		storeKey:            storeKey,
+		clobKeeper:          clobKeeper,
+		perpetualsKeeper:    perpetualsKeeper,
+		pricesKeeper:        pricesKeeper,
+		sendingKeeper:       sendingKeeper,
+		subaccountsKeeper:   subaccountsKeeper,
+		indexerEventManager: indexerEventManager,
+		authorities:         lib.UniqueSliceToSet(authorities),
 	}
 }
 
+func (k Keeper) GetIndexerEventManager() indexer_manager.IndexerEventManager {
+	return k.indexerEventManager
+}
+
 func (k Keeper) HasAuthority(authority string) bool {
 	_, ok := k.authorities[authority]
 	return ok

protocol/x/vault/keeper/orders.go

@@ -7,6 +7,8 @@ import (
 
 	errorsmod "cosmossdk.io/errors"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	indexerevents "github.com/dydxprotocol/v4-chain/protocol/indexer/events"
+	"github.com/dydxprotocol/v4-chain/protocol/indexer/indexer_manager"
 	"github.com/dydxprotocol/v4-chain/protocol/lib"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/log"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
@@ -25,6 +27,7 @@ func (k Keeper) RefreshAllVaultOrders(ctx sdk.Context) {
 	defer totalSharesIterator.Close()
 	for ; totalSharesIterator.Valid(); totalSharesIterator.Next() {
 		vaultId, err := types.GetVaultIdFromStateKey(totalSharesIterator.Key())
+
 		if err != nil {
 			log.ErrorLogWithError(ctx, "Failed to get vault ID from state key", err)
 			continue
@@ -85,7 +88,7 @@ func (k Keeper) RefreshVaultClobOrders(ctx sdk.Context, vaultId types.VaultId) (
 			err := k.clobKeeper.HandleMsgCancelOrder(ctx, clobtypes.NewMsgCancelOrderStateful(
 				order.OrderId,
 				uint32(ctx.BlockTime().Unix())+orderExpirationSeconds,
-			))
+			), true)
 			if err != nil {
 				log.ErrorLogWithError(ctx, "Failed to cancel order", err, "order", order, "vaultId", vaultId)
 			}
@@ -95,24 +98,52 @@ func (k Keeper) RefreshVaultClobOrders(ctx sdk.Context, vaultId types.VaultId) (
 			)
 		}
 	}
-
 	// Place new CLOB orders.
 	ordersToPlace, err := k.GetVaultClobOrders(ctx, vaultId)
 	if err != nil {
 		log.ErrorLogWithError(ctx, "Failed to get vault clob orders to place", err, "vaultId", vaultId)
 		return err
 	}
-	for _, order := range ordersToPlace {
+
+	for i, order := range ordersToPlace {
 		err := k.PlaceVaultClobOrder(ctx, order)
 		if err != nil {
 			log.ErrorLogWithError(ctx, "Failed to place order", err, "order", order, "vaultId", vaultId)
 		}
+
 		vaultId.IncrCounterWithLabels(
 			metrics.VaultPlaceOrder,
 			metrics.GetLabelForBoolValue(metrics.Success, err == nil),
 		)
-	}
 
+		// Send indexer messages. We expect ordersToCancel and ordersToPlace to have the same length
+		// and the order to place at each index to be a replacement of the order to cancel at the same index.
+		replacedOrder := ordersToCancel[i]
+		if replacedOrder == nil {
+			k.GetIndexerEventManager().AddTxnEvent(
+				ctx,
+				indexerevents.SubtypeStatefulOrder,
+				indexerevents.StatefulOrderEventVersion,
+				indexer_manager.GetBytes(
+					indexerevents.NewLongTermOrderPlacementEvent(
+						*order,
+					),
+				),
+			)
+		} else {
+			k.GetIndexerEventManager().AddTxnEvent(
+				ctx,
+				indexerevents.SubtypeStatefulOrder,
+				indexerevents.StatefulOrderEventVersion,
+				indexer_manager.GetBytes(
+					indexerevents.NewLongTermOrderReplacementEvent(
+						replacedOrder.OrderId,
+						*order,
+					),
+				),
+			)
+		}
+	}
 	return nil
 }