[CT-875] Simplify indexer messages for short term order replacement

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

@@ -641,12 +641,6 @@ describe('order-place-handler', () => {
         testConstants.defaultPerpetualMarket,
         APIOrderStatusEnum.BEST_EFFORT_OPENED,
         expectSubaccountMessage,
-        expectOrderBookUpdate
-          ? OrderbookMessage.fromPartial({
-            contents: JSON.stringify(orderbookContents),
-            clobPairId: testConstants.defaultPerpetualMarket.clobPairId,
-            version: ORDERBOOKS_WEBSOCKET_MESSAGE_VERSION,
-          }) : undefined,
       );
       expectStats(true);
     });

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

@@ -99,7 +99,7 @@ export class OrderPlaceHandler extends Handler {
 
     // TODO(IND-68): Remove once order replacement flow in V4 protocol removes the old order and
     // places the updated order.
-    const updatedQuantums: number | undefined = await this.updatePriceLevel(
+    await this.updatePriceLevel(
       placeOrderResult,
       perpetualMarket,
       update,
@@ -157,20 +157,6 @@ export class OrderPlaceHandler extends Handler {
       };
       sendMessageWrapper(subaccountMessage, KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS);
     }
-
-    // TODO(IND-68): Remove once order replacement flow in V4 protocol removes the old order and
-    // places the updated order.
-    if (updatedQuantums !== undefined) {
-      const orderbookMessage: Message = {
-        value: this.createOrderbookWebsocketMessage(
-          placeOrderResult.oldOrder!,
-          perpetualMarket,
-          updatedQuantums,
-        ),
-        headers,
-      };
-      sendMessageWrapper(orderbookMessage, KafkaTopics.TO_WEBSOCKETS_ORDERBOOKS);
-    }
   }
 
   /**

protocol/x/clob/memclob/memclob.go

@@ -495,23 +495,28 @@ func (m *MemClobPriceTimePriority) PlaceOrder(
 	}
 
 	if m.generateOffchainUpdates {
-		// If this is a replacement order, then ensure we send the appropriate replacement message.
+		// Send an order place message.
+		// If existing order ID is found and the price of the existing order is different from the new order,
+		// create an order removal message first so we can remove the original price level from the orderbook.
 		orderId := order.OrderId
-		if _, found := orderbook.getOrder(orderId); found {
-			if message, success := off_chain_updates.CreateOrderReplaceMessage(
-				ctx,
-				order,
-			); success {
-				offchainUpdates.AddReplaceMessage(orderId, message)
-			}
-		} else {
-			if message, success := off_chain_updates.CreateOrderPlaceMessage(
-				ctx,
-				order,
-			); success {
-				offchainUpdates.AddPlaceMessage(order.OrderId, message)
+		if existingOrder, found := orderbook.getOrder(orderId); found {
+			if order.Subticks != existingOrder.Subticks {
+				if message, success := off_chain_updates.CreateOrderRemoveMessageWithReason(
+					ctx,
+					orderId,
+					indexersharedtypes.OrderRemovalReason_ORDER_REMOVAL_REASON_REPLACED,
+					ocutypes.OrderRemoveV1_ORDER_REMOVAL_STATUS_BEST_EFFORT_CANCELED,
+				); success {
+					offchainUpdates.AddRemoveMessage(orderId, message)
+				}
 			}
 		}
+		if message, success := off_chain_updates.CreateOrderPlaceMessage(
+			ctx,
+			order,
+		); success {
+			offchainUpdates.AddPlaceMessage(order.OrderId, message)
+		}
 	}
 
 	// Attempt to match the order against the orderbook.

protocol/x/clob/memclob/memclob_place_order_reduce_only_test.go

@@ -1385,6 +1385,7 @@ func TestPlaceOrder_ReduceOnly(t *testing.T) {
 				tc.expectedCancelledReduceOnlyOrders,
 				// TODO(IND-261): Add tests for replaced reduce-only orders.
 				false,
+				false,
 			)
 		})
 	}
@@ -1668,6 +1669,7 @@ func TestPlaceOrder_LongTermReduceOnlyRemovals(t *testing.T) {
 				tc.expectedNewMatches,
 				tc.expectedCancelledReduceOnlyOrders,
 				false,
+				false,
 			)
 		})
 	}

protocol/x/clob/memclob/memclob_place_order_test.go

@@ -319,6 +319,7 @@ func TestPlaceOrder_AddOrderToOrderbook(t *testing.T) {
 				ordersOnBook = append(ordersOnBook, &order)
 			}
 
+			expectedReplacementOrderPriceChanged := false
 			for _, matchableOrder := range ordersOnBook {
 				// Note we assume these are regular orders since liquidation orders cannot rest on
 				// the book.
@@ -327,6 +328,9 @@ func TestPlaceOrder_AddOrderToOrderbook(t *testing.T) {
 				// is no longer on the book.
 				matchableOrderOrder := matchableOrder.MustGetOrder()
 				if matchableOrderOrder.OrderId == tc.order.OrderId && tc.order.MustCmpReplacementOrder(&matchableOrderOrder) > 0 {
+					if matchableOrderOrder.Subticks != tc.order.Subticks {
+						expectedReplacementOrderPriceChanged = true
+					}
 					continue
 				}
 
@@ -378,6 +382,7 @@ func TestPlaceOrder_AddOrderToOrderbook(t *testing.T) {
 				[]expectedMatch{},
 				[]types.OrderId{},
 				tc.expectedToReplaceOrder,
+				expectedReplacementOrderPriceChanged,
 			)
 		})
 	}
@@ -2215,8 +2220,7 @@ func TestPlaceOrder_MatchOrders_PreexistingMatches(t *testing.T) {
 
 			expectedOrderStatus:    types.InternalError,
 			expectedToReplaceOrder: true,
-
-			expectedErr: types.ErrFokOrderCouldNotBeFullyFilled,
+			expectedErr:            types.ErrFokOrderCouldNotBeFullyFilled,
 		},
 	}
 
@@ -2275,6 +2279,7 @@ func TestPlaceOrder_MatchOrders_PreexistingMatches(t *testing.T) {
 				tc.expectedNewMatches,
 				[]types.OrderId{},
 				tc.expectedToReplaceOrder,
+				false,
 			)
 		})
 	}
@@ -3246,6 +3251,7 @@ func TestPlaceOrder_PostOnly(t *testing.T) {
 				[]expectedMatch{},
 				[]types.OrderId{},
 				false,
+				false,
 			)
 		})
 	}
@@ -3389,6 +3395,7 @@ func TestPlaceOrder_ImmediateOrCancel(t *testing.T) {
 				tc.expectedCollatCheck,
 				[]types.OrderId{},
 				false,
+				false,
 			)
 		})
 	}
@@ -4063,6 +4070,7 @@ func TestPlaceOrder_FillOrKill(t *testing.T) {
 				expectedMatches,
 				[]types.OrderId{},
 				false,
+				false,
 			)
 		})
 	}

protocol/x/clob/memclob/memclob_test_util.go

@@ -1292,28 +1292,35 @@ func assertPlaceOrderOffchainMessages(
 	expectedNewMatches []expectedMatch,
 	expectedCancelledReduceOnlyOrders []types.OrderId,
 	expectedToReplaceOrder bool,
+	expectedReplacementOrderPriceChanged bool,
 ) {
 	actualOffchainMessages := offchainUpdates.GetMessages()
 	expectedOffchainMessages := []msgsender.Message{}
 	seenCancelledReduceOnlyOrders := mapset.NewSet[types.OrderId]()
 
 	// If there are no errors expected, an order place message should be sent.
 	if expectedErr == nil || doesErrorProduceOffchainMessages(expectedErr) {
-		var updateMessage msgsender.Message
 		if expectedToReplaceOrder {
-			updateMessage = off_chain_updates.MustCreateOrderReplaceMessage(
-				ctx,
-				order,
-			)
-		} else {
-			updateMessage = off_chain_updates.MustCreateOrderPlaceMessage(
-				ctx,
-				order,
-			)
+			if expectedReplacementOrderPriceChanged {
+				removeMessage := off_chain_updates.MustCreateOrderRemoveMessageWithReason(
+					ctx,
+					order.OrderId,
+					indexershared.OrderRemovalReason_ORDER_REMOVAL_REASON_REPLACED,
+					ocutypes.OrderRemoveV1_ORDER_REMOVAL_STATUS_BEST_EFFORT_CANCELED,
+				)
+				expectedOffchainMessages = append(
+					expectedOffchainMessages,
+					removeMessage,
+				)
+			}
 		}
+		placeMessage := off_chain_updates.MustCreateOrderPlaceMessage(
+			ctx,
+			order,
+		)
 		expectedOffchainMessages = append(
 			expectedOffchainMessages,
-			updateMessage,
+			placeMessage,
 		)
 		require.Equal(t, expectedOffchainMessages, actualOffchainMessages[:len(expectedOffchainMessages)])
 	}