Pin orders to Mesh coming from trusted staking pools (#157)

* Remove unused method

* Initial pass at pinning orders from trusted staking pools

* Fix linter error

* Improve comment

* Move pinned pool IDs and MMer address to configs

* Prettier fix

.env_example

@@ -19,3 +19,5 @@ HTTP_KEEP_ALIVE_TIMEOUT=
 HTTP_HEADERS_TIMEOUT=
 SRA_ORDER_EXPIRATION_BUFFER_SECONDS=
 MAX_EXPIRATION_ALERT_THRESHOLD_SECONDS=
+PINNED_POOL_IDS=
+PINNED_MM_ADDRESSES=

README.md

@@ -75,6 +75,8 @@ To get a local development version of `0x-api` running:
 | `WHITELIST_ALL_TOKENS`            | `false`                                                         | A boolean determining whether all tokens should be allowed to be posted.                                                                                                                                  |
 | `MESH_IGNORED_ADDRESSES`          | `[]`                                                            | A comma seperated list of addresses to ignore. These addresses are ignored at the ingress (Mesh) layer and are never persisted                                                                            |
 | `SWAP_IGNORED_ADDRESSES`          | `[]`                                                            | A comma seperated list of addresses to ignore. These addresses are persisted but not used in any `/swap/*` endpoints                                                                                      |
+| `PINNED_POOL_IDS`                 | `[]`                                                            | A comma seperated list of pool IDs whose MMers orders will be pinned to the Mesh node. This makes them immune to spam attacks.                                                                            |
+| `PINNED_MM_ADDRESSES`             | `[]`                                                            | A comma seperated list of MMer addresses whose orders will be pinned to the Mesh node. This makes them immune to spam attacks.                                                                            |
 
 3. Install the dependencies:
 

src/config.ts

@@ -17,6 +17,7 @@ import { ChainId } from './types';
 
 enum EnvVarType {
     AddressList,
+    StringList,
     Port,
     KeepAliveTimeout,
     ChainId,
@@ -68,6 +69,16 @@ export const SWAP_IGNORED_ADDRESSES: string[] = _.isEmpty(process.env.SWAP_IGNOR
     ? []
     : assertEnvVarType('SWAP_IGNORED_ADDRESSES', process.env.SWAP_IGNORED_ADDRESSES, EnvVarType.AddressList);
 
+// MMer addresses whose orders should be pinned to the Mesh node
+export const PINNED_POOL_IDS: string[] = _.isEmpty(process.env.PINNED_POOL_IDS)
+    ? []
+    : assertEnvVarType('PINNED_POOL_IDS', process.env.PINNED_POOL_IDS, EnvVarType.StringList);
+
+// MMer addresses whose orders should be pinned to the Mesh node
+export const PINNED_MM_ADDRESSES: string[] = _.isEmpty(process.env.PINNED_MM_ADDRESSES)
+    ? []
+    : assertEnvVarType('PINNED_MM_ADDRESSES', process.env.PINNED_MM_ADDRESSES, EnvVarType.AddressList);
+
 // Ethereum RPC Url
 export const ETHEREUM_RPC_URL = assertEnvVarType('ETHEREUM_RPC_URL', process.env.ETHEREUM_RPC_URL, EnvVarType.Url);
 
@@ -237,6 +248,10 @@ function assertEnvVarType(name: string, value: any, expectedType: EnvVarType): a
             const addressList = (value as string).split(',').map(a => a.toLowerCase());
             addressList.forEach((a, i) => assert.isETHAddressHex(`${name}[${i}]`, a));
             return addressList;
+        case EnvVarType.StringList:
+            assert.isString(name, value);
+            const stringList = (value as string).split(',');
+            return stringList;
         case EnvVarType.WhitelistAllTokens:
             return '*';
         case EnvVarType.FeeAssetData:

src/handlers/sra_handlers.ts

@@ -75,7 +75,9 @@ export class SRAHandlers {
             validateAssetDataIsWhitelistedOrThrow(allowedTokens, signedOrder.makerAssetData, 'makerAssetData');
             validateAssetDataIsWhitelistedOrThrow(allowedTokens, signedOrder.takerAssetData, 'takerAssetData');
         }
-        await this._orderBook.addOrderAsync(signedOrder);
+        const pinResult = await this._orderBook.splitOrdersByPinningAsync([signedOrder]);
+        const isPinned = pinResult.pin.length === 1;
+        await this._orderBook.addOrderAsync(signedOrder, isPinned);
         res.status(HttpStatus.OK).send();
     }
     public async postOrdersAsync(req: express.Request, res: express.Response): Promise<void> {
@@ -88,7 +90,11 @@ export class SRAHandlers {
                 validateAssetDataIsWhitelistedOrThrow(allowedTokens, signedOrder.takerAssetData, 'takerAssetData');
             }
         }
-        await this._orderBook.addOrdersAsync(signedOrders);
+        const pinResult = await this._orderBook.splitOrdersByPinningAsync(signedOrders);
+        await Promise.all([
+            this._orderBook.addOrdersAsync(pinResult.pin, true),
+            this._orderBook.addOrdersAsync(pinResult.doNotPin, false),
+        ]);
         res.status(HttpStatus.OK).send();
     }
 }

src/order_book_service_order_provider.ts

@@ -1,4 +1,4 @@
-import { AssetPairsItem, BaseOrderProvider, OrderStore, SignedOrder } from '@0x/asset-swapper';
+import { AcceptedRejectedOrders, AssetPairsItem, BaseOrderProvider, OrderStore, SignedOrder } from '@0x/asset-swapper';
 
 import { FIRST_PAGE } from './constants';
 import { OrderBookService } from './services/orderbook_service';
@@ -26,7 +26,7 @@ export class OrderBookServiceOrderProvider extends BaseOrderProvider {
     public async destroyAsync(): Promise<void> {
         return Promise.resolve();
     }
-    public async addOrdersAsync(_orders: SignedOrder[]): Promise<any> {
+    public async addOrdersAsync(_orders: SignedOrder[]): Promise<AcceptedRejectedOrders> {
         return Promise.resolve() as any;
     }
 }

src/services/order_watcher_service.ts

@@ -30,7 +30,14 @@ export class OrderWatcherService {
         // TODO(dekz): Mesh can reject due to InternalError or EthRPCRequestFailed.
         // in the future we can attempt to retry these a few times. Ultimately if we
         // cannot validate the order we cannot keep the order around
-        const { accepted, rejected } = await this._meshClient.addOrdersAsync(signedOrders);
+        const pinResult = await orderUtils.splitOrdersByPinningAsync(this._connection, signedOrders);
+        const [pinnedValidationResults, unpinnedValidationResults] = await Promise.all([
+            this._meshClient.addOrdersAsync(pinResult.pin, true),
+            this._meshClient.addOrdersAsync(pinResult.doNotPin, false),
+        ]);
+        const accepted = [...pinnedValidationResults.accepted, ...unpinnedValidationResults.accepted];
+        const rejected = [...pinnedValidationResults.rejected, ...unpinnedValidationResults.rejected];
+
         logger.info('OrderWatcherService sync', {
             accepted: accepted.length,
             rejected: rejected.length,

src/services/orderbook_service.ts

@@ -7,6 +7,7 @@ import { SRA_ORDER_EXPIRATION_BUFFER_SECONDS } from '../config';
 import { SignedOrderEntity } from '../entities';
 import { ValidationError } from '../errors';
 import { alertOnExpiredOrders } from '../logger';
+import { PinResult } from '../types';
 import { MeshClient } from '../utils/mesh_client';
 import { meshUtils } from '../utils/mesh_utils';
 import { orderUtils } from '../utils/order_utils';
@@ -154,12 +155,12 @@ export class OrderBookService {
         this._meshClient = meshClient;
         this._connection = connection;
     }
-    public async addOrderAsync(signedOrder: SignedOrder): Promise<void> {
-        return this.addOrdersAsync([signedOrder]);
+    public async addOrderAsync(signedOrder: SignedOrder, pinned: boolean): Promise<void> {
+        return this.addOrdersAsync([signedOrder], pinned);
     }
-    public async addOrdersAsync(signedOrders: SignedOrder[]): Promise<void> {
+    public async addOrdersAsync(signedOrders: SignedOrder[], pinned: boolean): Promise<void> {
         if (this._meshClient) {
-            const { rejected } = await this._meshClient.addOrdersAsync(signedOrders);
+            const { rejected } = await this._meshClient.addOrdersAsync(signedOrders, pinned);
             if (rejected.length !== 0) {
                 const validationErrors = rejected.map((r, i) => ({
                     field: `signedOrder[${i}]`,
@@ -173,4 +174,7 @@ export class OrderBookService {
         }
         throw new Error('Could not add order to mesh.');
     }
+    public async splitOrdersByPinningAsync(signedOrders: SignedOrder[]): Promise<PinResult> {
+        return orderUtils.splitOrdersByPinningAsync(this._connection, signedOrders);
+    }
 }

src/types.ts

@@ -387,3 +387,8 @@ export interface GetSwapQuoteResponseLiquiditySource {
     name: string;
     proportion: BigNumber;
 }
+
+export interface PinResult {
+    pin: SignedOrder[];
+    doNotPin: SignedOrder[];
+}

src/utils/order_store_db_adapter.ts

@@ -1,4 +1,4 @@
-import { AddedRemovedOrders, APIOrder, OrderSet, OrderStore } from '@0x/asset-swapper';
+import { APIOrder, OrderSet, OrderStore } from '@0x/asset-swapper';
 
 import { SWAP_IGNORED_ADDRESSES } from '../config';
 import { FIRST_PAGE } from '../constants';
@@ -33,13 +33,6 @@ export class OrderStoreDbAdapter extends OrderStore {
         await orderSet.addManyAsync(allowedOrders);
         return orderSet;
     }
-    public async updateAsync(addedRemoved: AddedRemovedOrders): Promise<void> {
-        const { added } = addedRemoved;
-        for (const order of added) {
-            await this._orderbookService.addOrderAsync(order.order);
-        }
-        // Currently not handling deletes as this is handled by Mesh
-    }
     public async getBatchOrderSetsForAssetsAsync(
         makerAssetDatas: string[],
         takerAssetDatas: string[],

src/utils/order_utils.ts

@@ -15,20 +15,24 @@ import {
     StaticCallAssetData,
 } from '@0x/types';
 import { BigNumber, errorUtils } from '@0x/utils';
+import { Connection } from 'typeorm';
 
 import {
     CHAIN_ID,
     DEFAULT_ERC20_TOKEN_PRECISION,
     FEE_RECIPIENT_ADDRESS,
     MAKER_FEE_ASSET_DATA,
     MAKER_FEE_UNIT_AMOUNT,
+    PINNED_MM_ADDRESSES,
+    PINNED_POOL_IDS,
     SRA_ORDER_EXPIRATION_BUFFER_SECONDS,
     TAKER_FEE_ASSET_DATA,
     TAKER_FEE_UNIT_AMOUNT,
 } from '../config';
 import { MAX_TOKEN_SUPPLY_POSSIBLE, NULL_ADDRESS, ONE_SECOND_MS } from '../constants';
 import { SignedOrderEntity } from '../entities';
-import { APIOrderWithMetaData } from '../types';
+import * as queries from '../queries/staking_queries';
+import { APIOrderWithMetaData, PinResult, RawEpochPoolStats } from '../types';
 
 const DEFAULT_ERC721_ASSET = {
     minAmount: new BigNumber(0),
@@ -278,4 +282,29 @@ export const orderUtils = {
         }
         return filteredOrders;
     },
+    // splitOrdersByPinning splits the orders into those we wish to pin in our Mesh node and
+    // those we wish not to pin. We wish to pin the orders of MMers with a lot of ZRX at stake and
+    // who have a track record of acting benevolently.
+    async splitOrdersByPinningAsync(connection: Connection, signedOrders: SignedOrder[]): Promise<PinResult> {
+        const currentPoolStats = await connection.query(queries.currentEpochPoolsStatsQuery);
+        let makerAddresses: string[] = PINNED_MM_ADDRESSES;
+        currentPoolStats.forEach((poolStats: RawEpochPoolStats) => {
+            if (!PINNED_POOL_IDS.includes(poolStats.pool_id)) {
+                return;
+            }
+            makerAddresses = [...makerAddresses, ...poolStats.maker_addresses];
+        });
+        const pinResult: PinResult = {
+            pin: [],
+            doNotPin: [],
+        };
+        signedOrders.forEach(signedOrder => {
+            if (makerAddresses.includes(signedOrder.makerAddress)) {
+                pinResult.pin.push(signedOrder);
+            } else {
+                pinResult.doNotPin.push(signedOrder);
+            }
+        });
+        return pinResult;
+    },
 };