OrderbookMidPrice Cache & Candles Fix

.github/mergify.yml

@@ -117,4 +117,20 @@ pull_request_rules:
       backport:
         branches:
           - release/protocol/v6.x
+  - name: backport to release/indexer/v7.x branch
+    conditions:
+      - base=main
+      - label=backport/indexer/v7.x
+    actions:
+      backport:
+        branches:
+          - release/indexer/v7.x
+  - name: backport to release/protocol/v7.x branch
+    conditions:
+      - base=main
+      - label=backport/protocol/v7.x
+    actions:
+      backport:
+        branches:
+          - release/protocol/v7.x
 

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

@@ -6,53 +6,14 @@ 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
+      - 'adam/candles-memory-cache'
     # TODO(DEC-837): Customize github build and push to ECR by service with paths
 
 jobs:
-  # Build and push to dev
-  call-build-and-push-ecs-services-dev:
-    name: (Dev) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev
-    secrets: inherit
-
   # Build and push to dev2
   call-build-and-push-ecs-services-dev2:
     name: (Dev2) Build and Push ECS Services
     uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
     with:
       ENVIRONMENT: dev2
-    secrets: inherit
-
-  # Build and push to dev3
-  call-build-and-push-ecs-services-dev3:
-    name: (Dev3) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev3
-    secrets: inherit
-
-  # Build and push to dev4
-  call-build-and-push-ecs-services-dev4:
-    name: (Dev4) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev4
-    secrets: inherit
-
-  # Build and push to dev5
-  call-build-and-push-ecs-services-dev5:
-    name: (Dev5) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev5
-    secrets: inherit
-
-  # Build and push to staging
-  call-build-and-push-ecs-services-staging:
-    name: (Staging) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: staging
-    secrets: inherit
+    secrets: inherit

.gitignore

@@ -8,6 +8,8 @@ v4-proto-py/v4_proto
 v4-proto-js/build
 v4-proto-js/node_modules
 v4-proto-js/src
+v4-proto-rs/target
+v4-proto-rs/Cargo.lock
 
 .idea
 .vscode

indexer/packages/base/package.json

@@ -32,6 +32,7 @@
   },
   "homepage": "https://github.com/dydxprotocol/indexer#readme",
   "dependencies": {
+    "@aws-sdk/client-ec2": "^3.354.0",
     "axios": "^1.2.1",
     "big.js": "^6.2.1",
     "bignumber.js": "^9.0.2",

indexer/packages/base/src/az-id.ts

@@ -0,0 +1,51 @@
+import { DescribeAvailabilityZonesCommand, EC2Client } from '@aws-sdk/client-ec2';
+
+import { axiosRequest } from './axios';
+import config from './config';
+import logger from './logger';
+
+export async function getAvailabilityZoneId(): Promise<string> {
+  if (config.ECS_CONTAINER_METADATA_URI_V4 !== '' && config.AWS_REGION !== '') {
+    const taskUrl = `${config.ECS_CONTAINER_METADATA_URI_V4}/task`;
+    try {
+      const response = await axiosRequest({
+        method: 'GET',
+        url: taskUrl,
+      }) as { AvailabilityZone: string };
+      const client = new EC2Client({ region: config.AWS_REGION });
+      const command = new DescribeAvailabilityZonesCommand({
+        ZoneNames: [response.AvailabilityZone],
+      });
+      try {
+        const ec2Response = await client.send(command);
+        const zoneId = ec2Response.AvailabilityZones![0].ZoneId!;
+        logger.info({
+          at: 'az-id#getAvailabilityZoneId',
+          message: `Got availability zone id ${zoneId}.`,
+        });
+        return ec2Response.AvailabilityZones![0].ZoneId!;
+      } catch (error) {
+        logger.error({
+          at: 'az-id#getAvailabilityZoneId',
+          message: 'Failed to fetch availabilty zone id from EC2. ',
+          error,
+        });
+        return '';
+      }
+    } catch (error) {
+      logger.error({
+        at: 'az-id#getAvailabilityZoneId',
+        message: 'Failed to retrieve availability zone from metadata endpoint. No availabilty zone id found.',
+        error,
+        taskUrl,
+      });
+      return '';
+    }
+  } else {
+    logger.error({
+      at: 'az-id#getAvailabilityZoneId',
+      message: 'No metadata URI or region. No availabilty zone id found.',
+    });
+    return '';
+  }
+}

indexer/packages/base/src/config.ts

@@ -39,6 +39,7 @@ export const baseConfigSchema = {
   STATSD_PORT: parseInteger({ default: 8125 }),
   LOG_LEVEL: parseString({ default: 'debug' }),
   ECS_CONTAINER_METADATA_URI_V4: parseString({ default: '' }),
+  AWS_REGION: parseString({ default: '' }),
 };
 
 export default parseSchema(baseConfigSchema);

indexer/packages/base/src/index.ts

@@ -14,6 +14,7 @@ export * from './bugsnag';
 export * from './stats-util';
 export * from './date-helpers';
 export * from './instance-id';
+export * from './az-id';
 
 // Do this outside logger.ts to avoid a dependency cycle with logger transports that may trigger
 // additional logging.

indexer/packages/compliance/src/clients/elliptic-provider.ts

@@ -26,6 +26,8 @@ export const API_PATH: string = '/v2/wallet/synchronous';
 export const API_URI: string = `https://aml-api.elliptic.co${API_PATH}`;
 export const RISK_SCORE_KEY: string = 'risk_score';
 export const NO_RULES_TRIGGERED_RISK_SCORE: number = -1;
+// We use different negative values of risk score to represent different elliptic response states
+export const NOT_IN_BLOCKCHAIN_RISK_SCORE: number = -2;
 
 export class EllipticProviderClient extends ComplianceClient {
   private apiKey: string;
@@ -98,7 +100,7 @@ export class EllipticProviderClient extends ComplianceClient {
           `${config.SERVICE_NAME}.get_elliptic_risk_score.status_code`,
           { status: '404' },
         );
-        return NO_RULES_TRIGGERED_RISK_SCORE;
+        return NOT_IN_BLOCKCHAIN_RISK_SCORE;
       }
 
       if (error?.response?.status === 429) {

indexer/packages/compliance/src/index.ts

@@ -5,3 +5,4 @@ export * from './geoblocking/util';
 export * from './types';
 export * from './config';
 export * from './constants';
+export * from './clients/elliptic-provider';

indexer/packages/kafka/__tests__/consumer.test.ts

@@ -10,10 +10,10 @@ import { TO_ENDER_TOPIC } from '../src';
 describe.skip('consumer', () => {
   beforeAll(async () => {
     await Promise.all([
-      consumer.connect(),
+      consumer!.connect(),
       producer.connect(),
     ]);
-    await consumer.subscribe({ topic: TO_ENDER_TOPIC });
+    await consumer!.subscribe({ topic: TO_ENDER_TOPIC });
     await startConsumer();
   });
 

indexer/packages/kafka/src/consumer.ts

@@ -1,4 +1,5 @@
 import {
+  getAvailabilityZoneId,
   logger,
 } from '@dydxprotocol-indexer/base';
 import {
@@ -13,15 +14,10 @@ const groupIdPrefix: string = config.SERVICE_NAME;
 const groupIdSuffix: string = config.KAFKA_ENABLE_UNIQUE_CONSUMER_GROUP_IDS ? `_${uuidv4()}` : '';
 const groupId: string = `${groupIdPrefix}${groupIdSuffix}`;
 
-export const consumer: Consumer = kafka.consumer({
-  groupId,
-  sessionTimeout: config.KAFKA_SESSION_TIMEOUT_MS,
-  rebalanceTimeout: config.KAFKA_REBALANCE_TIMEOUT_MS,
-  heartbeatInterval: config.KAFKA_HEARTBEAT_INTERVAL_MS,
-  maxWaitTimeInMs: config.KAFKA_WAIT_MAX_TIME_MS,
-  readUncommitted: false,
-  maxBytes: 4194304, // 4MB
-});
+// As a hack, we made this mutable since CommonJS doesn't support top level await.
+// Top level await would needed to fetch the az id (used as rack id).
+// eslint-disable-next-line import/no-mutable-exports
+export let consumer: Consumer | undefined;
 
 // List of functions to run per message consumed.
 let onMessageFunction: (topic: string, message: KafkaMessage) => Promise<void>;
@@ -51,38 +47,51 @@ export function updateOnBatchFunction(
 // Whether the consumer is stopped.
 let stopped: boolean = false;
 
-consumer.on('consumer.disconnect', async () => {
+export async function stopConsumer(): Promise<void> {
   logger.info({
-    at: 'consumers#disconnect',
-    message: 'Kafka consumer disconnected',
+    at: 'kafka-consumer#stop',
+    message: 'Stopping kafka consumer',
     groupId,
   });
 
-  if (!stopped) {
-    await consumer.connect();
-    logger.info({
-      at: 'kafka-consumer#disconnect',
-      message: 'Kafka consumer reconnected',
-      groupId,
-    });
-  } else {
+  stopped = true;
+  await consumer!.disconnect();
+}
+
+export async function initConsumer(): Promise<void> {
+  consumer = kafka.consumer({
+    groupId,
+    sessionTimeout: config.KAFKA_SESSION_TIMEOUT_MS,
+    rebalanceTimeout: config.KAFKA_REBALANCE_TIMEOUT_MS,
+    heartbeatInterval: config.KAFKA_HEARTBEAT_INTERVAL_MS,
+    maxWaitTimeInMs: config.KAFKA_WAIT_MAX_TIME_MS,
+    readUncommitted: false,
+    maxBytes: 4194304, // 4MB
+    rackId: await getAvailabilityZoneId(),
+  });
+
+  consumer!.on('consumer.disconnect', async () => {
     logger.info({
-      at: 'kafka-consumer#disconnect',
-      message: 'Not reconnecting since task is shutting down',
+      at: 'consumers#disconnect',
+      message: 'Kafka consumer disconnected',
       groupId,
     });
-  }
-});
 
-export async function stopConsumer(): Promise<void> {
-  logger.info({
-    at: 'kafka-consumer#stop',
-    message: 'Stopping kafka consumer',
-    groupId,
+    if (!stopped) {
+      await consumer!.connect();
+      logger.info({
+        at: 'kafka-consumer#disconnect',
+        message: 'Kafka consumer reconnected',
+        groupId,
+      });
+    } else {
+      logger.info({
+        at: 'kafka-consumer#disconnect',
+        message: 'Not reconnecting since task is shutting down',
+        groupId,
+      });
+    }
   });
-
-  stopped = true;
-  await consumer.disconnect();
 }
 
 export async function startConsumer(batchProcessing: boolean = false): Promise<void> {
@@ -104,7 +113,7 @@ export async function startConsumer(batchProcessing: boolean = false): Promise<v
     };
   }
 
-  await consumer.run(consumerRunConfig);
+  await consumer!.run(consumerRunConfig);
 
   logger.info({
     at: 'consumers#connect',

indexer/packages/postgres/__tests__/helpers/constants.ts

@@ -71,6 +71,7 @@ export const invalidTicker: string = 'INVALID-INVALID';
 export const dydxChain: string = 'dydx';
 export const defaultAddress: string = 'dydx1n88uc38xhjgxzw9nwre4ep2c8ga4fjxc565lnf';
 export const defaultAddress2: string = 'dydx1n88uc38xhjgxzw9nwre4ep2c8ga4fjxc575lnf';
+export const defaultAddress3: string = 'dydx199tqg4wdlnu4qjlxchpd7seg454937hjrknju4';
 export const blockedAddress: string = 'dydx1f9k5qldwmqrnwy8hcgp4fw6heuvszt35egvtx2';
 // Vault address for vault id 0 was generated using
 // script protocol/scripts/vault/get_vault.go
@@ -100,6 +101,20 @@ export const defaultSubaccount3: SubaccountCreateObject = {
   updatedAtHeight: createdHeight,
 };
 
+export const defaultSubaccount2Num0: SubaccountCreateObject = {
+  address: defaultAddress2,
+  subaccountNumber: 0,
+  updatedAt: createdDateTime.toISO(),
+  updatedAtHeight: createdHeight,
+};
+
+export const defaultSubaccount3Num0: SubaccountCreateObject = {
+  address: defaultAddress3,
+  subaccountNumber: 0,
+  updatedAt: createdDateTime.toISO(),
+  updatedAtHeight: createdHeight,
+};
+
 // defaultWalletAddress belongs to defaultWallet2 and is different from defaultAddress
 export const defaultSubaccountDefaultWalletAddress: SubaccountCreateObject = {
   address: defaultWalletAddress,
@@ -990,9 +1005,10 @@ export const defaultAffiliateInfo: AffiliateInfoCreateObject = {
   affiliateEarnings: '10',
   referredMakerTrades: 10,
   referredTakerTrades: 20,
-  totalReferredFees: '10',
+  totalReferredMakerFees: '10',
+  totalReferredTakerFees: '10',
+  totalReferredMakerRebates: '-10',
   totalReferredUsers: 5,
-  referredNetProtocolEarnings: '20',
   firstReferralBlockHeight: '1',
   referredTotalVolume: '1000',
 };
@@ -1002,9 +1018,10 @@ export const defaultAffiliateInfo2: AffiliateInfoCreateObject = {
   affiliateEarnings: '11',
   referredMakerTrades: 11,
   referredTakerTrades: 21,
-  totalReferredFees: '11',
+  totalReferredMakerFees: '11',
+  totalReferredTakerFees: '11',
+  totalReferredMakerRebates: '-11',
   totalReferredUsers: 5,
-  referredNetProtocolEarnings: '21',
   firstReferralBlockHeight: '11',
   referredTotalVolume: '1000',
 };
@@ -1014,9 +1031,10 @@ export const defaultAffiliateInfo3: AffiliateInfoCreateObject = {
   affiliateEarnings: '12',
   referredMakerTrades: 12,
   referredTakerTrades: 22,
-  totalReferredFees: '12',
+  totalReferredMakerFees: '12',
+  totalReferredTakerFees: '12',
+  totalReferredMakerRebates: '-12',
   totalReferredUsers: 10,
-  referredNetProtocolEarnings: '22',
   firstReferralBlockHeight: '12',
   referredTotalVolume: '1111111',
 };

indexer/packages/postgres/__tests__/helpers/mock-generators.ts

@@ -35,8 +35,17 @@ import {
   isolatedPerpetualMarket2,
   isolatedSubaccount,
   isolatedSubaccount2,
+  defaultSubaccount2Num0,
+  defaultSubaccount3Num0,
 } from './constants';
 
+export async function seedAdditionalSubaccounts() {
+  await Promise.all([
+    SubaccountTable.create(defaultSubaccount2Num0),
+    SubaccountTable.create(defaultSubaccount3Num0),
+  ]);
+}
+
 export async function seedData() {
   await Promise.all([
     SubaccountTable.create(defaultSubaccount),