[IND-460] Emit deleveraging events from protocol (#736)

indexer/packages/v4-protos/src/codegen/dydxprotocol/indexer/events/events.ts

@@ -444,6 +444,62 @@ export interface OrderFillEventV1SDKType {
 
   total_filled_taker: Long;
 }
+/**
+ * DeleveragingEvent message contains all the information for a deleveraging
+ * on the dYdX chain. This includes the liquidated/offsetting subaccounts and
+ * the amount filled.
+ */
+
+export interface DeleveragingEventV1 {
+  /** ID of the subaccount that was liquidated. */
+  liquidated?: IndexerSubaccountId;
+  /** ID of the subaccount that was used to offset the position. */
+
+  offsetting?: IndexerSubaccountId;
+  /** The ID of the perpetual that was liquidated. */
+
+  perpetualId: number;
+  /**
+   * The amount filled between the liquidated and offsetting position, in
+   * base quantums.
+   */
+
+  fillAmount: Long;
+  /** Bankruptcy price of liquidated subaccount, in USDC quote quantums. */
+
+  price: Long;
+  /** `true` if liquidating a short position, `false` otherwise. */
+
+  isBuy: boolean;
+}
+/**
+ * DeleveragingEvent message contains all the information for a deleveraging
+ * on the dYdX chain. This includes the liquidated/offsetting subaccounts and
+ * the amount filled.
+ */
+
+export interface DeleveragingEventV1SDKType {
+  /** ID of the subaccount that was liquidated. */
+  liquidated?: IndexerSubaccountIdSDKType;
+  /** ID of the subaccount that was used to offset the position. */
+
+  offsetting?: IndexerSubaccountIdSDKType;
+  /** The ID of the perpetual that was liquidated. */
+
+  perpetual_id: number;
+  /**
+   * The amount filled between the liquidated and offsetting position, in
+   * base quantums.
+   */
+
+  fill_amount: Long;
+  /** Bankruptcy price of liquidated subaccount, in USDC quote quantums. */
+
+  price: Long;
+  /** `true` if liquidating a short position, `false` otherwise. */
+
+  is_buy: boolean;
+}
 /**
  * LiquidationOrder represents the liquidation taker order to be included in a
  * liquidation order fill event.
@@ -1719,6 +1775,101 @@ export const OrderFillEventV1 = {
 
 };
 
+function createBaseDeleveragingEventV1(): DeleveragingEventV1 {
+  return {
+    liquidated: undefined,
+    offsetting: undefined,
+    perpetualId: 0,
+    fillAmount: Long.UZERO,
+    price: Long.UZERO,
+    isBuy: false
+  };
+}
+
+export const DeleveragingEventV1 = {
+  encode(message: DeleveragingEventV1, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+    if (message.liquidated !== undefined) {
+      IndexerSubaccountId.encode(message.liquidated, writer.uint32(10).fork()).ldelim();
+    }
+
+    if (message.offsetting !== undefined) {
+      IndexerSubaccountId.encode(message.offsetting, writer.uint32(18).fork()).ldelim();
+    }
+
+    if (message.perpetualId !== 0) {
+      writer.uint32(24).uint32(message.perpetualId);
+    }
+
+    if (!message.fillAmount.isZero()) {
+      writer.uint32(32).uint64(message.fillAmount);
+    }
+
+    if (!message.price.isZero()) {
+      writer.uint32(40).uint64(message.price);
+    }
+
+    if (message.isBuy === true) {
+      writer.uint32(48).bool(message.isBuy);
+    }
+
+    return writer;
+  },
+
+  decode(input: _m0.Reader | Uint8Array, length?: number): DeleveragingEventV1 {
+    const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
+    let end = length === undefined ? reader.len : reader.pos + length;
+    const message = createBaseDeleveragingEventV1();
+
+    while (reader.pos < end) {
+      const tag = reader.uint32();
+
+      switch (tag >>> 3) {
+        case 1:
+          message.liquidated = IndexerSubaccountId.decode(reader, reader.uint32());
+          break;
+
+        case 2:
+          message.offsetting = IndexerSubaccountId.decode(reader, reader.uint32());
+          break;
+
+        case 3:
+          message.perpetualId = reader.uint32();
+          break;
+
+        case 4:
+          message.fillAmount = (reader.uint64() as Long);
+          break;
+
+        case 5:
+          message.price = (reader.uint64() as Long);
+          break;
+
+        case 6:
+          message.isBuy = reader.bool();
+          break;
+
+        default:
+          reader.skipType(tag & 7);
+          break;
+      }
+    }
+
+    return message;
+  },
+
+  fromPartial(object: DeepPartial<DeleveragingEventV1>): DeleveragingEventV1 {
+    const message = createBaseDeleveragingEventV1();
+    message.liquidated = object.liquidated !== undefined && object.liquidated !== null ? IndexerSubaccountId.fromPartial(object.liquidated) : undefined;
+    message.offsetting = object.offsetting !== undefined && object.offsetting !== null ? IndexerSubaccountId.fromPartial(object.offsetting) : undefined;
+    message.perpetualId = object.perpetualId ?? 0;
+    message.fillAmount = object.fillAmount !== undefined && object.fillAmount !== null ? Long.fromValue(object.fillAmount) : Long.UZERO;
+    message.price = object.price !== undefined && object.price !== null ? Long.fromValue(object.price) : Long.UZERO;
+    message.isBuy = object.isBuy ?? false;
+    return message;
+  }
+
+};
+
 function createBaseLiquidationOrderV1(): LiquidationOrderV1 {
   return {
     liquidated: undefined,

indexer/services/ender/__tests__/helpers/constants.ts

@@ -26,7 +26,11 @@ import {
   OrderRemovalReason,
   AssetCreateEventV1,
   PerpetualMarketCreateEventV1,
-  ClobPairStatus, LiquidityTierUpsertEventV1, UpdatePerpetualEventV1, UpdateClobPairEventV1,
+  ClobPairStatus,
+  LiquidityTierUpsertEventV1,
+  UpdatePerpetualEventV1,
+  UpdateClobPairEventV1,
+  DeleveragingEventV1,
 } from '@dydxprotocol-indexer/v4-protos';
 import Long from 'long';
 import { DateTime } from 'luxon';
@@ -276,6 +280,14 @@ export const defaultTransferEvent: TransferEventV1 = {
     subaccountId: defaultRecipientSubaccountId,
   },
 };
+export const defaultDeleveragingEvent: DeleveragingEventV1 = {
+  liquidated: defaultSenderSubaccountId,
+  offsetting: defaultRecipientSubaccountId,
+  perpetualId: 1,
+  fillAmount: Long.fromValue(10_000, true),
+  price: Long.fromValue(1_000_000_000, true),
+  isBuy: true,
+};
 export const defaultDepositEvent: TransferEventV1 = {
   assetId: 0,
   amount: Long.fromValue(100, true),

indexer/services/ender/__tests__/lib/on-message.test.ts

@@ -18,6 +18,7 @@ import {
   TransactionTable,
 } from '@dydxprotocol-indexer/postgres';
 import {
+  DeleveragingEventV1,
   FundingEventV1,
   IndexerTendermintBlock,
   IndexerTendermintEvent,
@@ -39,6 +40,7 @@ import { logger, stats } from '@dydxprotocol-indexer/base';
 import { TransferHandler } from '../../src/handlers/transfer-handler';
 import { FundingHandler } from '../../src/handlers/funding-handler';
 import {
+  defaultDeleveragingEvent,
   defaultFundingUpdateSampleEvent,
   defaultHeight,
   defaultMarketModify,
@@ -49,10 +51,12 @@ import { updateBlockCache } from '../../src/caches/block-cache';
 import { MarketModifyHandler } from '../../src/handlers/markets/market-modify-handler';
 import Long from 'long';
 import { createPostgresFunctions } from '../../src/helpers/postgres/postgres-functions';
+import { DeleveragingHandler } from '../../src/handlers/deleveraging-handler';
 
 jest.mock('../../src/handlers/subaccount-update-handler');
 jest.mock('../../src/handlers/transfer-handler');
 jest.mock('../../src/handlers/funding-handler');
+jest.mock('../../src/handlers/deleveraging-handler');
 jest.mock('../../src/handlers/markets/market-modify-handler');
 
 describe('on-message', () => {
@@ -80,6 +84,11 @@ describe('on-message', () => {
       validate: () => null,
       getParallelizationIds: () => [],
     });
+    (DeleveragingHandler as jest.Mock).mockReturnValue({
+      handle: () => [],
+      validate: () => null,
+      getParallelizationIds: () => [],
+    });
     producerSendMock = jest.spyOn(producer, 'send');
     producerSendMock.mockImplementation(() => {
     });
@@ -153,6 +162,10 @@ describe('on-message', () => {
     defaultMarketModify,
   ).finish());
 
+  const defaultDeleveragingEventBinary: Uint8Array = Uint8Array.from(DeleveragingEventV1.encode(
+    defaultDeleveragingEvent,
+  ).finish());
+
   it.each([
     [
       'via knex',
@@ -369,6 +382,63 @@ describe('on-message', () => {
       expect.any(Number), 1, { success: 'true' });
   });
 
+  it('successfully processes block with deleveraging event', async () => {
+    await Promise.all([
+      MarketTable.create(testConstants.defaultMarket),
+      MarketTable.create(testConstants.defaultMarket2),
+    ]);
+    await Promise.all([
+      LiquidityTiersTable.create(testConstants.defaultLiquidityTier),
+      LiquidityTiersTable.create(testConstants.defaultLiquidityTier2),
+    ]);
+    await Promise.all([
+      PerpetualMarketTable.create(testConstants.defaultPerpetualMarket),
+      PerpetualMarketTable.create(testConstants.defaultPerpetualMarket2),
+    ]);
+    await perpetualMarketRefresher.updatePerpetualMarkets();
+
+    const transactionIndex: number = 0;
+    const eventIndex: number = 0;
+    const events: IndexerTendermintEvent[] = [
+      createIndexerTendermintEvent(
+        DydxIndexerSubtypes.DELEVERAGING,
+        defaultDeleveragingEventBinary,
+        transactionIndex,
+        eventIndex,
+      ),
+    ];
+
+    const block: IndexerTendermintBlock = createIndexerTendermintBlock(
+      defaultHeight,
+      defaultTime,
+      events,
+      [defaultTxHash],
+    );
+    const binaryBlock: Uint8Array = Uint8Array.from(IndexerTendermintBlock.encode(block).finish());
+    const kafkaMessage: KafkaMessage = createKafkaMessage(Buffer.from(binaryBlock));
+
+    await onMessage(kafkaMessage);
+    await Promise.all([
+      expectTendermintEvent(defaultHeight.toString(), transactionIndex, eventIndex),
+      expectBlock(defaultHeight.toString(), defaultDateTime.toISO()),
+    ]);
+
+    expect((DeleveragingHandler as jest.Mock)).toHaveBeenCalledTimes(1);
+    expect((DeleveragingHandler as jest.Mock)).toHaveBeenNthCalledWith(
+      1,
+      block,
+      events[0],
+      expect.any(Number),
+      defaultDeleveragingEvent,
+    );
+    expect(stats.increment).toHaveBeenCalledWith('ender.received_kafka_message', 1);
+    expect(stats.timing).toHaveBeenCalledWith(
+      'ender.message_time_in_queue', expect.any(Number), 1, { topic: KafkaTopics.TO_ENDER });
+    expect(stats.gauge).toHaveBeenCalledWith('ender.processing_block_height', expect.any(Number));
+    expect(stats.timing).toHaveBeenCalledWith('ender.processed_block.timing',
+      expect.any(Number), 1, { success: 'true' });
+  });
+
   it('throws error while processing unparsable messages', async () => {
     const transactionIndex: number = 0;
     const eventIndex: number = 0;

indexer/services/ender/__tests__/validators/deleveraging-validator.test.ts

@@ -0,0 +1,80 @@
+import { logger, ParseMessageError } from '@dydxprotocol-indexer/base';
+import { DeleveragingEventV1, IndexerTendermintBlock, IndexerTendermintEvent } from '@dydxprotocol-indexer/v4-protos';
+import { DydxIndexerSubtypes } from '../../src/lib/types';
+import { DeleveragingValidator } from '../../src/validators/deleveraging-validator';
+import {
+  defaultDeleveragingEvent, defaultHeight, defaultTime, defaultTxHash,
+} from '../helpers/constants';
+import { createIndexerTendermintBlock, createIndexerTendermintEvent } from '../helpers/indexer-proto-helpers';
+import { expectDidntLogError, expectLoggedParseMessageError } from '../helpers/validator-helpers';
+
+describe('deleveraging-validator', () => {
+  beforeEach(() => {
+    jest.spyOn(logger, 'error');
+  });
+
+  afterEach(() => {
+    jest.clearAllMocks();
+  });
+
+  describe('validate', () => {
+    it('does not throw error on valid deleveraging', () => {
+      const validator: DeleveragingValidator = new DeleveragingValidator(
+        defaultDeleveragingEvent,
+        createBlock(defaultDeleveragingEvent),
+      );
+
+      validator.validate();
+      expectDidntLogError();
+    });
+
+    it.each([
+      [
+        'does not contain liquidated',
+        {
+          ...defaultDeleveragingEvent,
+          liquidated: undefined,
+        },
+        'DeleveragingEvent must have a liquidated subaccount id',
+      ],
+      [
+        'does not contain offsetting',
+        {
+          ...defaultDeleveragingEvent,
+          offsetting: undefined,
+        },
+        'DeleveragingEvent must have an offsetting subaccount id',
+      ],
+    ])('throws error if event %s', (_message: string, event: DeleveragingEventV1, message: string) => {
+      const validator: DeleveragingValidator = new DeleveragingValidator(
+        event,
+        createBlock(event),
+      );
+
+      expect(() => validator.validate()).toThrow(new ParseMessageError(message));
+      expectLoggedParseMessageError(
+        DeleveragingValidator.name,
+        message,
+        { event },
+      );
+    });
+  });
+});
+
+function createBlock(
+  deleveragingEvent: DeleveragingEventV1,
+): IndexerTendermintBlock {
+  const event: IndexerTendermintEvent = createIndexerTendermintEvent(
+    DydxIndexerSubtypes.DELEVERAGING,
+    DeleveragingEventV1.encode(deleveragingEvent).finish(),
+    0,
+    0,
+  );
+
+  return createIndexerTendermintBlock(
+    defaultHeight,
+    defaultTime,
+    [event],
+    [defaultTxHash],
+  );
+}

indexer/services/ender/src/constants.ts

@@ -13,3 +13,7 @@ export const SUBACCOUNT_ORDER_FILL_EVENT_TYPE: string = 'subaccount_order_fill';
 
 // StatefulOrder and OrderFill events for the same order are processed chronologically.
 export const STATEFUL_ORDER_ORDER_FILL_EVENT_TYPE: string = 'stateful_order_order_fill';
+
+// Deleveraging, SubaccountUpdate, and OrderFill events for the same subaccount
+// are processed chronologically.
+export const DELEVERAGING_EVENT_TYPE: string = 'deleveraging';