Add QA to ETH transfers exporter

src/blockchains/eth/eth_types.ts

@@ -75,9 +75,9 @@ export type ETHTransfer = {
   valueExactBase36: string,
   blockNumber: number,
   timestamp: number,
-  transactionHash?: string,
-  transactionPosition?: number,
-  internalTxPosition?: number,
+  transactionHash: string,
+  transactionPosition: number,
+  internalTxPosition: number,
   type: string,
   primaryKey?: number,
 }

src/blockchains/eth/eth_worker.ts

@@ -2,7 +2,7 @@ import { logger } from '../../lib/logger';
 import { constructRPCClient } from '../../lib/http_client';
 import { injectDAOHackTransfers, DAO_HACK_FORK_BLOCK } from './lib/dao_hack';
 import { getGenesisTransfers } from './lib/genesis_transfers';
-import { assignInternalTransactionPosition, transactionOrder } from './lib/util'
+import { assignInternalTransactionPosition, doQAETHTransfers, transactionOrder, mergeSortedArrays } from './lib/util'
 import { BaseWorker } from '../../lib/worker_base';
 import { Web3Interface, constructWeb3Wrapper, safeCastToNumber } from './lib/web3_wrapper';
 import { decodeTransferTrace } from './lib/decode_transfers';
@@ -105,13 +105,15 @@ export class ETHWorker extends BaseWorker {
     logger.info(`Fetching transfer events for interval ${fromBlock}:${toBlock}`)
     const [traces, blocks, receipts] = await this.fetchData(fromBlock, toBlock)
     const events: (ETHTransfer | EOB)[] = this.transformPastEvents(fromBlock, toBlock, traces, blocks, receipts)
-    assignInternalTransactionPosition(events)
-    events.push(...collectEndOfBlocks(fromBlock, toBlock, blocks, this.web3Wrapper))
     events.sort(transactionOrder)
+    doQAETHTransfers(events, fromBlock, toBlock)
+    assignInternalTransactionPosition(events)
+    const eobEvents = collectEndOfBlocks(fromBlock, toBlock, blocks, this.web3Wrapper)
+    const mergedEvents = mergeSortedArrays(events, eobEvents, transactionOrder)
 
     this.lastExportedBlock = toBlock
 
-    return events
+    return mergedEvents
   }
 
   async init(): Promise<void> {

src/blockchains/eth/lib/decode_transfers.ts

@@ -18,6 +18,7 @@ export function decodeTransferTrace(trace: Trace, timestamp: number, web3Wrapper
       timestamp: timestamp,
       transactionHash: trace['transactionHash'],
       transactionPosition: trace['transactionPosition'],
+      internalTxPosition: 0,
       type: trace['type']
     };
   }
@@ -37,6 +38,7 @@ export function decodeTransferTrace(trace: Trace, timestamp: number, web3Wrapper
       timestamp: timestamp,
       transactionHash: trace['transactionHash'],
       transactionPosition: trace['transactionPosition'],
+      internalTxPosition: 0,
       type: trace['type']
     };
   }
@@ -55,6 +57,7 @@ export function decodeTransferTrace(trace: Trace, timestamp: number, web3Wrapper
       timestamp: timestamp,
       transactionHash: trace['transactionHash'],
       transactionPosition: trace['transactionPosition'],
+      internalTxPosition: 0,
       type: trace['type']
     };
   }
@@ -76,6 +79,7 @@ export function decodeTransferTrace(trace: Trace, timestamp: number, web3Wrapper
     timestamp: timestamp,
     transactionHash: trace['transactionHash'],
     transactionPosition: trace['transactionPosition'],
+    internalTxPosition: 0,
     type: trace['type']
   };
 }

src/blockchains/eth/lib/end_of_block.ts

@@ -10,6 +10,7 @@ export type EOB = {
     timestamp: number,
     transactionHash: string,
     transactionPosition: number,
+    internalTxPosition: number,
     type: string,
     primaryKey?: number
 };

src/blockchains/eth/lib/fees_decoder.ts

@@ -26,6 +26,8 @@ export class FeesDecoder {
       blockNumber: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.blockNumber)),
       timestamp: safeCastToNumber(this.web3Wrapper.parseHexToNumber(block.timestamp)),
       transactionHash: transaction.hash,
+      transactionPosition: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.transactionIndex)),
+      internalTxPosition: 0,
       type: 'fee'
     }];
   }
@@ -46,6 +48,8 @@ export class FeesDecoder {
       blockNumber: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.blockNumber)),
       timestamp: safeCastToNumber(this.web3Wrapper.parseHexToNumber(block.timestamp)),
       transactionHash: transaction.hash,
+      transactionPosition: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.transactionIndex)),
+      internalTxPosition: 0,
       type: 'fee_burnt'
     };
   }
@@ -75,6 +79,8 @@ export class FeesDecoder {
         blockNumber: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.blockNumber)),
         timestamp: safeCastToNumber(this.web3Wrapper.parseHexToNumber(block.timestamp)),
         transactionHash: transaction.hash,
+        transactionPosition: safeCastToNumber(this.web3Wrapper.parseHexToNumber(transaction.transactionIndex)),
+        internalTxPosition: 0,
         type: 'fee'
       };
     }

src/blockchains/eth/lib/genesis_transfers.ts

@@ -12,10 +12,17 @@ const GENESIS_TIMESTAMP = 1438269973;
 
 export function getGenesisTransfers(web3Wrapper: Web3Interface): ETHTransfer[] {
   const result: ETHTransfer[] = [];
+
+  const txHashMap: Map<string, number> = new Map();
+
   GENESIS_TRANSFERS.forEach((transfer) => {
     const [id, from, to, amount] = transfer;
     const wei = web3Wrapper.etherToWei(amount);
 
+    // Used to construct incrementing internal transaction numbers
+    const currentCount = txHashMap.get(from) || 0;
+
+
     result.push({
       from: 'GENESIS',
       to: to,
@@ -24,6 +31,8 @@ export function getGenesisTransfers(web3Wrapper: Web3Interface): ETHTransfer[] {
       blockNumber: 0,
       timestamp: GENESIS_TIMESTAMP,
       transactionHash: from,
+      transactionPosition: currentCount,
+      internalTxPosition: 0,
       type: 'genesis'
     });
   });
@@ -35,6 +44,9 @@ export function getGenesisTransfers(web3Wrapper: Web3Interface): ETHTransfer[] {
     valueExactBase36: BigInt('5000000000000000000').toString(36),
     blockNumber: 0,
     timestamp: GENESIS_TIMESTAMP,
+    transactionHash: 'GENESIS_mining_tx',
+    transactionPosition: 0,
+    internalTxPosition: 0,
     type: 'reward'
   });
 

src/blockchains/eth/lib/util.ts

@@ -1,7 +1,8 @@
 import { ETHTransfer } from '../eth_types';
+import { EOB } from './end_of_block'
 const { groupBy } = require('lodash');
 
-export function transactionOrder(a: ETHTransfer, b: ETHTransfer) {
+export function transactionOrder(a: ETHTransfer | EOB, b: ETHTransfer | EOB) {
   if (a.blockNumber !== b.blockNumber) {
     return a.blockNumber - b.blockNumber
   }
@@ -21,15 +22,85 @@ export function transactionOrder(a: ETHTransfer, b: ETHTransfer) {
 
 const ethTransferKey = (transfer: ETHTransfer) => `${transfer.blockNumber}-${transfer.transactionHash ?? ''}-${transfer.transactionPosition ?? ''}-${transfer.from}-${transfer.to}`
 
-export function assignInternalTransactionPosition(transfers: ETHTransfer[], groupByKey: (transfer: ETHTransfer) => string = ethTransferKey): ETHTransfer[] {
+export function assignInternalTransactionPosition(transfers: ETHTransfer[], groupByKey: (transfer: ETHTransfer) => string = ethTransferKey): void {
   const grouped = groupBy(transfers, groupByKey)
   const values: ETHTransfer[][] = Object.values(grouped)
 
-  return values.flatMap((transfersSameKey: ETHTransfer[]) => {
+  values.forEach((transfersSameKey: ETHTransfer[]) => {
     transfersSameKey.forEach((transfer: ETHTransfer, index: number) => {
       transfer.internalTxPosition = index
     })
-    return transfersSameKey
   })
 }
 
+export function assertBlocksMatch(groupedTransfers: any, fromBlock: number, toBlock: number) {
+  const keys = Object.keys(groupedTransfers)
+  if (keys.length !== toBlock - fromBlock + 1) {
+    throw new Error(`Wrong number of blocks seen. Expected ${toBlock - fromBlock + 1} got ${keys.length}.`)
+  }
+
+  for (let block = fromBlock; block <= toBlock; block++) {
+    if (!groupedTransfers.hasOwnProperty(block.toString())) {
+      throw new Error(`Missing transfers for block ${block}.`)
+    }
+  }
+}
+
+
+export function assertTransfersWithinBlock(transfersPerBlock: ETHTransfer[]) {
+  let expectedTxPosition = 0
+
+  for (const transfer of transfersPerBlock) {
+    if (transfer.transactionPosition !== expectedTxPosition) {
+      // We allow for multiple transfers withing a transaction. That is why we can see the same transaction position several times.
+      if (transfer.transactionPosition !== expectedTxPosition + 1) {
+        throw new Error(`Unexpected transaction position for transfer: ${JSON.stringify(transfer)}, expected tx position: ${expectedTxPosition} or ${expectedTxPosition + 1}`);
+      }
+      expectedTxPosition += 1
+    }
+  }
+}
+
+
+/**
+ * Assert data quality guarantees on top of input Transfers
+ *
+ * Throw an error if:
+ * 1. A block number is missing
+ * 2. Within a block, a transactions number is missing
+ *
+ * @param transfers Ordered array of transfers
+ * @param fromBlock Block number indicating start of expected interval
+ * @param toBlock Block number indicating end of expected interval
+ */
+export function doQAETHTransfers(sortedTransfers: ETHTransfer[], fromBlock: number, toBlock: number) {
+  if (fromBlock > toBlock) {
+    throw new Error(`Invalid block range: fromBlock ${fromBlock} is greater than toBlock ${toBlock}`);
+  }
+
+  const groupedTransfers = groupBy(sortedTransfers, (transfer: ETHTransfer) => transfer.blockNumber)
+
+  assertBlocksMatch(groupedTransfers, fromBlock, toBlock);
+
+  for (const key of Object.keys(groupedTransfers)) {
+    assertTransfersWithinBlock(groupedTransfers[key])
+  }
+}
+
+export function mergeSortedArrays<T>(sortedArr1: T[], sortedArr2: T[], comparator: (a: T, b: T) => number = (a, b) =>
+  a < b ? -1 : a > b ? 1 : 0): T[] {
+  const merged: T[] = [];
+  let i = 0, j = 0;
+
+  while (i < sortedArr1.length && j < sortedArr2.length) {
+    if (comparator(sortedArr1[i], sortedArr2[j]) < 0) {
+      merged.push(sortedArr1[i++]);
+    } else {
+      merged.push(sortedArr2[j++]);
+    }
+  }
+
+  return merged.concat(sortedArr1.slice(i), sortedArr2.slice(j));
+}
+
+

src/blockchains/eth/lib/withdrawals_decoder.ts

@@ -10,7 +10,7 @@ export class WithdrawalsDecoder {
   }
 
   getBeaconChainWithdrawals(withdrawals: BeaconChainWithdrawal[], blockNumber: number, blockTimestamp: number): ETHTransfer[] {
-    return withdrawals.map((withdrawal) => {
+    return withdrawals.map((withdrawal, index) => {
       const gweiAmount = this.web3Wrapper.gweiToWei(withdrawal.amount);
       return {
         from: constants.ETH_WITHDRAWAL,
@@ -20,6 +20,8 @@ export class WithdrawalsDecoder {
         blockNumber: blockNumber,
         timestamp: blockTimestamp,
         transactionHash: `WITHDRAWAL_${blockNumber}`,
+        transactionPosition: 0,
+        internalTxPosition: index,
         type: 'beacon_withdrawal'
       };
     });

src/test/eth/decode_transfers.spec.ts

@@ -3,6 +3,7 @@ process.env.IS_ETH = 'true';
 import { decodeTransferTrace } from '../../blockchains/eth/lib/decode_transfers';
 import { NODE_URL } from '../../blockchains/eth/lib/constants';
 import { Web3Interface, constructWeb3WrapperNoCredentials } from '../../blockchains/eth/lib/web3_wrapper';
+import { ETHTransfer } from '../../blockchains/eth/eth_types';
 
 
 describe('genesis transfers', function () {
@@ -31,14 +32,15 @@ describe('genesis transfers', function () {
     const timestamp = 1000000;
     const result = decodeTransferTrace(suicide_trace, timestamp, web3Wrapper);
 
-    const result_expected = {
+    const result_expected: ETHTransfer = {
       'from': '0xa6c3b7f6520a0ef594fc666d3874ec78c561cdbb',
       'to': '0x245133ea0fb1b77fab5886d7ffb8046dfeff3858',
       'value': 160000000000000000,
       'valueExactBase36': '17rf9la2f4sg',
       'blockNumber': 711983,
       'timestamp': 1000000,
       'transactionHash': '0xd715da4f846e41be86ea87dc97b186cafea3b50c95d5d9d889ec522b248b207f',
+      'internalTxPosition': 0,
       'transactionPosition': 10,
       'type': 'suicide'
     };
@@ -74,7 +76,7 @@ describe('genesis transfers', function () {
 
     const result = decodeTransferTrace(call_trace, timestamp, web3Wrapper);
 
-    const result_expected = {
+    const result_expected: ETHTransfer = {
       'from': '0x48f2e6e5d0872da169c7f5823d5a2d5ea5f2b5e7',
       'to': '0x7de5aba7de728950c92c57d08e20d4077161f12f',
       'value': 1,
@@ -83,6 +85,7 @@ describe('genesis transfers', function () {
       'timestamp': 1450433505,
       'transactionHash': '0x22f839c82ff455554ec8aa98ee2b9a03d0d5ed4707b46d4a0a217df7d58bda2c',
       'transactionPosition': 10,
+      'internalTxPosition': 0,
       'type': 'call'
     };
 
@@ -112,17 +115,18 @@ describe('genesis transfers', function () {
     const timestamp = 1450433505;
 
     const result = decodeTransferTrace(reward_trace, timestamp, web3Wrapper);
-    const result_expected = {
+    const result_expected: ETHTransfer = {
       'from': 'mining_block',
       'to': '0x2a65aca4d5fc5b5c859090a6c34d164135398226',
       'transactionHash': '0x22f839c82ff455554ec8aa98ee2b9a03d0d5ed4707b46d4a0a217df7d58bda2c',
       'transactionPosition': 10,
+      'internalTxPosition': 0,
       'value': 5000000000000000000,
       'valueExactBase36': '11zk02pzlmwow',
       'blockNumber': 710093,
       'timestamp': 1450433505,
       'type': 'reward'
-    };
+    }
 
     assert.deepStrictEqual(result, result_expected);
   });
@@ -150,7 +154,7 @@ describe('genesis transfers', function () {
     const timestamp = 1450435908;
     const result = decodeTransferTrace(create_trace, timestamp, web3Wrapper);
 
-    const result_expected = {
+    const result_expected: ETHTransfer = {
       'from': '0x245133ea0fb1b77fab5886d7ffb8046dfeff3858',
       'to': '0xa6c3b7f6520a0ef594fc666d3874ec78c561cdbb',
       'value': 1500000000000000000,
@@ -159,8 +163,9 @@ describe('genesis transfers', function () {
       'timestamp': 1450435908,
       'transactionHash': '0x6d39df3c46f19e8ef5e8bb3b81a063a29cb352675a00d66f0dc2117a1799add1',
       'transactionPosition': 11,
+      'internalTxPosition': 0,
       'type': 'create'
-    };
+    }
 
     assert.deepStrictEqual(result, result_expected);
   });