Add better error handling for immediately invoked async functions

packages/pipeline/src/scripts/merge_v2_events.ts

@@ -1,81 +0,0 @@
-import { web3Factory } from '@0x/dev-utils';
-import 'reflect-metadata';
-import { Connection, createConnection } from 'typeorm';
-
-import { ExchangeEventsSource } from '../data_sources/contract-wrappers/exchange_events';
-import { ExchangeFillEvent } from '../entities';
-import { deployConfig } from '../ormconfig';
-import { parseExchangeEvents } from '../parsers/events';
-
-let connection: Connection;
-
-(async () => {
-    connection = await createConnection(deployConfig);
-    await getExchangeEventsAsync();
-    await mergeExchangeEventsAsync();
-    console.log('Exiting process');
-    process.exit(0);
-})();
-
-// TODO(albrow): Separately: Errors do not appear to be handled correctly. If you use the
-// wrong rpcUrl it just returns early with no error.
-async function getExchangeEventsAsync(): Promise<void> {
-    console.log('Getting event logs...');
-    const provider = web3Factory.getRpcProvider({
-        rpcUrl: 'https://mainnet.infura.io',
-    });
-    const eventsRepository = connection.getRepository(ExchangeFillEvent);
-    const exchangeEvents = new ExchangeEventsSource(provider, 1);
-    const eventLogs = await exchangeEvents.getFillEventsAsync();
-    console.log('Parsing events...');
-    const events = parseExchangeEvents(eventLogs);
-    console.log(`Retrieved and parsed ${events.length} total events.`);
-    console.log('Saving events...');
-    for (const event of events) {
-        await eventsRepository.save(event);
-    }
-    await eventsRepository.save(events);
-    console.log('Saved events.');
-}
-
-const insertEventsRawQuery = `INSERT INTO events_raw (
-    event_type,
-    error_id,
-    order_hash,
-    maker,
-    maker_amount,
-    maker_fee,
-    maker_token,
-    taker,
-    taker_amount,
-    taker_fee,
-    taker_token,
-    txn_hash,
-    fee_recipient,
-    block_number,
-    log_index
-)
-(
-    SELECT
-        'LogFill',
-        null,
-        "orderHash",
-        "makerAddress",
-        "makerAssetFilledAmount"::numeric(78),
-        "makerFeePaid"::numeric(78),
-        "makerTokenAddress",
-        "takerAddress",
-        "takerAssetFilledAmount"::numeric(78),
-        "takerFeePaid"::numeric(78),
-        "takerTokenAddress",
-        "transactionHash",
-        "feeRecipientAddress",
-        "blockNumber",
-        "logIndex"
-    FROM exchange_fill_event
-) ON CONFLICT (order_hash, txn_hash, log_index) DO NOTHING`;
-
-async function mergeExchangeEventsAsync(): Promise<void> {
-    console.log('Merging results into events_raw...');
-    await connection.query(insertEventsRawQuery);
-}

packages/pipeline/src/scripts/pull_missing_events.ts

@@ -8,6 +8,7 @@ import { ExchangeEventsSource } from '../data_sources/contract-wrappers/exchange
 import { ExchangeFillEvent } from '../entities';
 import { deployConfig } from '../ormconfig';
 import { parseExchangeEvents } from '../parsers/events';
+import { handleError } from '../utils';
 
 const EXCHANGE_START_BLOCK = 6271590; // Block number when the Exchange contract was deployed to mainnet.
 const START_BLOCK_OFFSET = 1000; // Number of blocks before the last known block to consider when updating fill events.
@@ -22,7 +23,7 @@ let connection: Connection;
     });
     await getExchangeEventsAsync(provider);
     process.exit(0);
-})();
+})().catch(handleError);
 
 async function getExchangeEventsAsync(provider: Web3ProviderEngine): Promise<void> {
     console.log('Checking existing event logs...');
@@ -53,8 +54,8 @@ async function getStartBlockAsync(eventsRepository: Repository<ExchangeFillEvent
         return EXCHANGE_START_BLOCK;
     }
     const queryResult = await connection.query(
-        'SELECT "blockNumber" FROM exchange_fill_event ORDER BY "blockNumber" DESC LIMIT 1',
+        'SELECT block_number FROM exchange_fill_events ORDER BY block_number DESC LIMIT 1',
     );
-    const lastKnownBlock = queryResult[0].blockNumber;
+    const lastKnownBlock = queryResult[0].block_number;
     return lastKnownBlock - START_BLOCK_OFFSET;
 }

packages/pipeline/src/scripts/update_relayer_info.ts

@@ -5,6 +5,7 @@ import { RelayerRegistrySource } from '../data_sources/relayer-registry';
 import { Relayer } from '../entities';
 import { deployConfig } from '../ormconfig';
 import { parseRelayers } from '../parsers/relayer_registry';
+import { handleError } from '../utils';
 
 // NOTE(albrow): We need to manually update this URL for now. Fix this when we
 // have the relayer-registry behind semantic versioning.
@@ -17,7 +18,7 @@ let connection: Connection;
     connection = await createConnection(deployConfig);
     await getRelayers();
     process.exit(0);
-})();
+})().catch(handleError);
 
 async function getRelayers(): Promise<void> {
     console.log('Getting latest relayer info...');

packages/pipeline/src/utils/index.ts

@@ -6,3 +6,17 @@ export function bigNumbertoStringOrNull(n: BigNumber): string | null {
     }
     return n.toString();
 }
+
+export function handleError(e: any): void {
+    if (e.message != null) {
+        console.error(e.message);
+    } else {
+        console.error('Unknown error');
+    }
+    if (e.stack != null) {
+        console.error(e.stack);
+    } else {
+        console.error('(No stack trace)');
+    }
+    process.exit(1);
+}