Fix vault orderbook flickering

.github/workflows/indexer-build-and-push-dev-staging.yml

@@ -6,6 +6,7 @@ on: # yamllint disable-line rule:truthy
       - main
       - 'release/indexer/v[0-9]+.[0-9]+.x'  # e.g. release/indexer/v0.1.x
       - 'release/indexer/v[0-9]+.x'  # e.g. release/indexer/v1.x
+      - 'chenyao/fix-vault-orderbook-flickering'
     # TODO(DEC-837): Customize github build and push to ECR by service with paths
 
 jobs:

.github/workflows/protocol-build-and-push.yml

@@ -6,6 +6,7 @@ on:  # yamllint disable-line rule:truthy
       - main
       - 'release/protocol/v[0-9]+.[0-9]+.x'  # e.g. release/protocol/v0.1.x
       - 'release/protocol/v[0-9]+.x'  # e.g. release/protocol/v1.x
+      - 'chenyao/fix-vault-orderbook-flickering'
 
 jobs:
   build-and-push-dev:

indexer/packages/redis/src/scripts/place_order.lua

@@ -66,7 +66,7 @@ else
   -- to "false".
   redis.call("set", orderKey, newOrder)
   -- refer to the above comment on order data format
-  redis.call("set", orderDataKey, newOrderExpiry .. "_" .. oldTotalFilledQuantums .. "_false")
+  redis.call("set", orderDataKey, newOrderExpiry .. "_" .. oldTotalFilledQuantums .. "_true")
   -- Long-term orders will be on-chain, so we only need to store expiry data for short-term orders
   if isShortTermOrder then
     -- The expiry is guaranteed to be different, so overwrite the old one from the expiry cache

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

@@ -536,9 +536,14 @@ describe('order-place-handler', () => {
     ) => {
       const oldOrderTotalFilled: number = 10;
       const oldPriceLevelInitialQuantums: number = Number(initialOrderToPlace.quantums) * 2;
+      // After replacing the order the quantums at the price level of the old order should be:
+      // initial quantums - (old order quantums - old order total filled)
+      const expectedPriceLevelQuantums: number = (
+        oldPriceLevelInitialQuantums - (Number(initialOrderToPlace.quantums) - oldOrderTotalFilled)
+      );
       const expectedPriceLevel: PriceLevel = {
         humanPrice: expectedRedisOrder.price,
-        quantums: oldPriceLevelInitialQuantums.toString(),
+        quantums: expectedPriceLevelQuantums.toString(),
         lastUpdated: expect.stringMatching(/^[0-9]{10}$/),
       };
 
@@ -1082,6 +1087,11 @@ describe('order-place-handler', () => {
         APIOrderStatusEnum.BEST_EFFORT_OPENED,
         true,
       );
+
+      expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({
+        at: 'OrderPlaceHandler#handle',
+        message: 'Total filled of order in Redis exceeds order quantums.',
+      }));
     });
   });
 });

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

@@ -6,19 +6,22 @@ import {
   OrderTable,
   PerpetualMarketFromDatabase,
   perpetualMarketRefresher,
+  protocolTranslations,
 } from '@dydxprotocol-indexer/postgres';
 import {
   CanceledOrdersCache,
   placeOrder,
   PlaceOrderResult,
   StatefulOrderUpdatesCache,
   convertToRedisOrder,
+  OrderbookLevelsCache,
 } from '@dydxprotocol-indexer/redis';
 import {
   getOrderIdHash,
   isLongTermOrder,
   isStatefulOrder,
   ORDER_FLAG_SHORT_TERM,
+  requiresImmediateExecution,
 } from '@dydxprotocol-indexer/v4-proto-parser';
 import {
   IndexerOrder,
@@ -28,6 +31,7 @@ import {
   OrderUpdateV1,
   RedisOrder,
 } from '@dydxprotocol-indexer/v4-protos';
+import Big from 'big.js';
 import { IHeaders, Message } from 'kafkajs';
 
 import config from '../config';
@@ -96,6 +100,12 @@ export class OrderPlaceHandler extends Handler {
       stats.increment(`${config.SERVICE_NAME}.place_order_handler.replaced_order`, 1);
     }
 
+    await this.updatePriceLevel(
+      placeOrderResult,
+      perpetualMarket,
+      update,
+    );
+
     // TODO(CLOB-597): Remove this logic and log erorrs once best-effort-open is not sent for
     // stateful orders in the protocol
     if (this.shouldSendSubaccountMessage(
@@ -158,6 +168,83 @@ export class OrderPlaceHandler extends Handler {
     }
   }
 
+  /**
+   * Updates the price level given the result of calling `placeOrder`.
+   * @param result `PlaceOrderResult` from calling `placeOrder`
+   * @param perpetualMarket Perpetual market object corresponding to the perpetual market of the
+   * order
+   * @param update Off-chain update
+   * @returns
+   */
+  // eslint-disable-next-line @typescript-eslint/require-await
+  protected async updatePriceLevel(
+    result: PlaceOrderResult,
+    perpetualMarket: PerpetualMarketFromDatabase,
+    update: OffChainUpdateV1,
+  ): Promise<number | undefined> {
+    // TODO(DEC-1339): Update price levels based on if the order is reduce-only and if the replaced
+    // order is reduce-only.
+    if (
+      result.replaced !== true ||
+      result.restingOnBook !== true ||
+      requiresImmediateExecution(result.oldOrder!.order!.timeInForce)
+    ) {
+      return undefined;
+    }
+
+    const remainingSizeDeltaInQuantums: Big = this.getRemainingSizeDeltaInQuantums(result);
+
+    if (remainingSizeDeltaInQuantums.eq(0)) {
+      // No update to the price level if remaining quantums = 0
+      // An order could have remaining quantums = 0 intra-block, as an order is only considered
+      // filled once the fills are committed in a block
+      return undefined;
+    }
+
+    if (remainingSizeDeltaInQuantums.lt(0)) {
+      // Log error and skip updating orderbook levels if old order had negative remaining
+      // quantums
+      logger.info({
+        at: 'OrderPlaceHandler#handle',
+        message: 'Total filled of order in Redis exceeds order quantums.',
+        placeOrderResult: result,
+        update,
+      });
+      stats.increment(`${config.SERVICE_NAME}.order_place_total_filled_exceeds_size`, 1);
+      return undefined;
+    }
+
+    // If the remaining size is not equal or less than 0, it must be greater than 0.
+    // Remove the remaining size of the replaced order from the orderbook, by decrementing
+    // the total size of orders at the price of the replaced order
+    return runFuncWithTimingStat(
+      OrderbookLevelsCache.updatePriceLevel({
+        ticker: perpetualMarket.ticker,
+        side: protocolTranslations.protocolOrderSideToOrderSide(result.oldOrder!.order!.side),
+        humanPrice: result.oldOrder!.price,
+        // Delta should be -1 * remaining size of order in quantums and an integer
+        sizeDeltaInQuantums: remainingSizeDeltaInQuantums.mul(-1).toFixed(0),
+        client: redisClient,
+      }),
+      this.generateTimingStatsOptions('update_price_level'),
+    );
+  }
+
+  /**
+   * Gets the remaining size of the old order if the order was replaced.
+   * @param result Result of placing an order, should be for a replaced order so both `oldOrder` and
+   * `oldTotalFilledQuantums` properties should exist on the place order result.
+   * @returns Remaining size of the old order that was replaced.
+   */
+  protected getRemainingSizeDeltaInQuantums(result: PlaceOrderResult): Big {
+    const sizeDeltaInQuantums: Big = Big(
+      result.oldOrder!.order!.quantums.toString(),
+    ).minus(
+      result.oldTotalFilledQuantums!,
+    );
+    return sizeDeltaInQuantums;
+  }
+
   /**
    * Determine whether to send a subaccount websocket message given the order place.
    * @param orderPlace

protocol/app/app.go

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

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)
 
@@ -111,17 +112,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 {
+		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

@@ -88,13 +88,15 @@ func (k Keeper) HandleMsgPlaceOrder(
 
 	// 2. Return an error if an associated cancellation or removal already exists in the current block.
 	processProposerMatchesEvents := k.GetProcessProposerMatchesEvents(ctx)
-	cancelledOrderIds := lib.UniqueSliceToSet(processProposerMatchesEvents.PlacedStatefulCancellationOrderIds)
-	if _, found := cancelledOrderIds[order.GetOrderId()]; found {
-		return errorsmod.Wrapf(
-			types.ErrStatefulOrderPreviouslyCancelled,
-			"PlaceOrder: order (%+v)",
-			order,
-		)
+	if !isInternalOrder { // If vault order, we allow the order to replace a cancelled order with the same order ID
+		cancelledOrderIds := lib.UniqueSliceToSet(processProposerMatchesEvents.PlacedStatefulCancellationOrderIds)
+		if _, found := cancelledOrderIds[order.GetOrderId()]; found {
+			return errorsmod.Wrapf(
+				types.ErrStatefulOrderPreviouslyCancelled,
+				"PlaceOrder: order (%+v)",
+				order,
+			)
+		}
 	}
 	removedOrderIds := lib.UniqueSliceToSet(processProposerMatchesEvents.RemovedStatefulOrderIds)
 	if _, found := removedOrderIds[order.GetOrderId()]; found {
@@ -115,31 +117,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 {
+			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 {
+			k.GetIndexerEventManager().AddTxnEvent(
+				ctx,
+				indexerevents.SubtypeStatefulOrder,
+				indexerevents.StatefulOrderEventVersion,
+				indexer_manager.GetBytes(
+					indexerevents.NewLongTermOrderPlacementEvent(
+						order,
+					),
 				),
-			),
-		)
+			)
+		}
 		processProposerMatchesEvents.PlacedLongTermOrderIds = append(
 			processProposerMatchesEvents.PlacedLongTermOrderIds,
 			order.OrderId,

protocol/x/clob/keeper/msg_server_place_order_test.go

@@ -435,13 +435,12 @@ func TestHandleMsgPlaceOrder(t *testing.T) {
 			removalExists:         false,
 			equityTierLimitExists: true,
 		},
-		"Error - Place an Internal Order, Order Already Cancelled": {
+		"Success - Place an Internal Order, Order Already Cancelled": {
 			isInternalOrder:       true,
 			assetQuantums:         -1_000_000_000,
 			cancellationExists:    true,
 			removalExists:         false,
 			equityTierLimitExists: true,
-			expectedError:         types.ErrStatefulOrderPreviouslyCancelled,
 		},
 		"Error - Place an Internal Order, Order Already Removed": {
 			isInternalOrder:       true,

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,