[Events Orderbook] E2e test for streamed orderbook. (#732)

This PR implements an e2e test for the streamed orderbook to verify that all events are applied and that the event based auction state matches the onchain orderbook.

### Test Plan

Run the e2e test:
```
$ yarn test-streamed-orderbook # This takes a LONG time
  Streamed Orderbook
    init
==> building streamed orderbook...
==> querying onchain orderbook...
==> comparing orderbooks...
      ✓ should successfully apply all events and match on-chain orderbook (338341ms)


  1 passing (6m)
```

src/encoding.js

@@ -18,6 +18,7 @@ class OrderBuffer {
 
     this.decodeAddr = () => `0x${this.readBytes(20)}`
     this.decodeInt = (size) => new BN(this.readBytes(size / 8), 16).toString()
+    this.decodeNumber = (size) => parseInt(this.readBytes(size / 8), 16)
 
     return this
   }
@@ -27,10 +28,10 @@ function decodeOrder(bytes) {
   return {
     user: bytes.decodeAddr(),
     sellTokenBalance: bytes.decodeInt(256),
-    buyToken: bytes.decodeInt(16),
-    sellToken: bytes.decodeInt(16),
-    validFrom: bytes.decodeInt(32),
-    validUntil: bytes.decodeInt(32),
+    buyToken: bytes.decodeNumber(16),
+    sellToken: bytes.decodeNumber(16),
+    validFrom: bytes.decodeNumber(32),
+    validUntil: bytes.decodeNumber(32),
     priceNumerator: bytes.decodeInt(128),
     priceDenominator: bytes.decodeInt(128),
     remainingAmount: bytes.decodeInt(128),
@@ -40,7 +41,7 @@ function decodeOrder(bytes) {
 function decodeIndexedOrder(bytes) {
   return {
     ...decodeOrder(bytes),
-    orderId: bytes.decodeInt(16),
+    orderId: bytes.decodeNumber(16),
   }
 }
 
@@ -72,12 +73,8 @@ function decodeOrders(bytes) {
 
 function decodeOrdersBN(bytes) {
   return decodeOrders(bytes).map((e) => ({
-    user: e.user,
+    ...e,
     sellTokenBalance: new BN(e.sellTokenBalance),
-    buyToken: parseInt(e.buyToken),
-    sellToken: parseInt(e.sellToken),
-    validFrom: parseInt(e.validFrom),
-    validUntil: parseInt(e.validUntil),
     priceNumerator: new BN(e.priceNumerator),
     priceDenominator: new BN(e.priceDenominator),
     remainingAmount: new BN(e.remainingAmount),
@@ -97,13 +94,8 @@ function decodeIndexedOrders(bytes) {
 
 function decodeIndexedOrdersBN(bytes) {
   return decodeIndexedOrders(bytes).map((e) => ({
-    user: e.user,
-    orderId: parseInt(e.orderId),
+    ...e,
     sellTokenBalance: new BN(e.sellTokenBalance),
-    buyToken: parseInt(e.buyToken),
-    sellToken: parseInt(e.sellToken),
-    validFrom: parseInt(e.validFrom),
-    validUntil: parseInt(e.validUntil),
     priceNumerator: new BN(e.priceNumerator),
     priceDenominator: new BN(e.priceDenominator),
     remainingAmount: new BN(e.remainingAmount),

src/index.d.ts

@@ -75,34 +75,33 @@ export interface ContractArtifact {
 export declare const BatchExchangeArtifact: ContractArtifact
 export declare const BatchExchangeViewerArtifact: ContractArtifact
 
-export interface Order {
+export interface Order<T = string> {
   user: string;
-  sellTokenBalance: string;
-  buyToken: string;
-  sellToken: string;
-  validFrom: string;
-  validUntil: string;
-  priceNumerator: string;
-  priceDenominator: string;
-  remainingAmount: string;
-}
-
-export interface OrderBN {
-  user: string;
-  sellTokenBalance: BN;
+  sellTokenBalance: T;
   buyToken: number;
   sellToken: number;
   validFrom: number;
   validUntil: number;
-  priceNumerator: BN;
-  priceDenominator: BN;
-  remainingAmount: BN;
+  priceNumerator: T;
+  priceDenominator: T;
+  remainingAmount: T;
 }
 
-export declare function decodeOrders(bytes: string): Order[];
-export declare function decodeOrdersBN(bytes: string): OrderBN[];
+export interface IndexedOrder<T> extends Order<T> {
+  orderId: number;
+}
+
+export declare function decodeOrders(bytes: string): Order<string>[];
+export declare function decodeOrdersBN(bytes: string): Order<BN>[];
 
 export declare function getOpenOrdersPaginated(
   contract: BatchExchangeViewer,
-  pageSize: number
-): AsyncIterable<OrderBN[]>;
+  pageSize: number,
+  blockNumber?: number | string,
+): AsyncIterable<IndexedOrder<BN>[]>;
+
+export declare function getOpenOrders(
+  contract: BatchExchangeViewer,
+  pageSize: number,
+  blockNumber?: number | string,
+): Promise<IndexedOrder<BN>[]>;

src/onchain_reading.js

@@ -4,14 +4,17 @@ const { decodeOrdersBN, decodeIndexedOrdersBN } = require("./encoding")
  * Returns an iterator yielding an item for each page of order in the orderbook that is currently being collected.
  * @param {BatchExchangeViewer} contract to query from
  * @param {number} pageSize the number of items to fetch per page
+ * @param {number?} blockNumber the block number to execute the query at, defaults to "latest" if omitted
  */
-const getOpenOrdersPaginated = async function* (contract, pageSize) {
+const getOpenOrdersPaginated = async function* (contract, pageSize, blockNumber) {
   let nextPageUser = "0x0000000000000000000000000000000000000000"
   let nextPageUserOffset = 0
   let hasNextPage = true
 
   while (hasNextPage) {
-    const page = await contract.methods.getOpenOrderBookPaginated([], nextPageUser, nextPageUserOffset, pageSize).call()
+    const page = await contract.methods
+      .getOpenOrderBookPaginated([], nextPageUser, nextPageUserOffset, pageSize)
+      .call({}, blockNumber)
     const elements = decodeIndexedOrdersBN(page.elements)
     yield elements
 
@@ -22,18 +25,35 @@ const getOpenOrdersPaginated = async function* (contract, pageSize) {
   }
 }
 
+/**
+ * Returns open orders in the orderbook.
+ * @param {BatchExchange} contract to query from
+ * @param {number} pageSize the number of items to fetch per page
+ * @param {number?} blockNumber the block number to execute the query at, defaults to "latest" if omitted
+ */
+const getOpenOrders = async (contract, pageSize, blockNumber) => {
+  let allOrders = []
+  for await (const page of getOpenOrdersPaginated(contract, pageSize, blockNumber)) {
+    allOrders = allOrders.concat(page)
+  }
+  return allOrders
+}
+
 /**
  * Returns all orders in the orderbook.
  * @param {BatchExchange} contract to query from
  * @param {number} pageSize the number of items to fetch per page
+ * @param {number?} blockNumber the block number to execute the query at, defaults to "latest" if omitted
  */
-const getOrdersPaginated = async (contract, pageSize) => {
+const getOrdersPaginated = async (contract, pageSize, blockNumber) => {
   let orders = []
   let currentUser = "0x0000000000000000000000000000000000000000"
   let currentOffSet = 0
   let lastPageSize = pageSize
   while (lastPageSize == pageSize) {
-    const page = decodeOrdersBN(await contract.methods.getEncodedUsersPaginated(currentUser, currentOffSet, pageSize).call())
+    const page = decodeOrdersBN(
+      await contract.methods.getEncodedUsersPaginated(currentUser, currentOffSet, pageSize).call({}, blockNumber)
+    )
     orders = orders.concat(page)
     for (const index in page) {
       if (page[index].user != currentUser) {
@@ -49,5 +69,6 @@ const getOrdersPaginated = async (contract, pageSize) => {
 
 module.exports = {
   getOpenOrdersPaginated,
+  getOpenOrders,
   getOrdersPaginated,
 }

src/streamed/index.ts

@@ -15,8 +15,9 @@
 
 import Web3 from "web3"
 import { BlockNumber, TransactionReceipt } from "web3-core"
+import { Contract } from "web3-eth-contract"
 import { AbiItem } from "web3-utils"
-import { BatchExchange, BatchExchangeArtifact } from "../.."
+import { BatchExchange, BatchExchangeArtifact, ContractArtifact, IndexedOrder } from ".."
 import { AnyEvent } from "./events"
 import { AuctionState } from "./state"
 
@@ -77,17 +78,13 @@ export const DEFAULT_ORDERBOOK_OPTIONS: OrderbookOptions = {
   strict: false,
 }
 
-/**
- * The duration in seconds of a batch.
- */
-export const BATCH_DURATION = 300
-
 /**
  * The streamed orderbook that manages incoming events, and applies them to the
  * account state.
  */
 export class StreamedOrderbook {
   private readonly state: AuctionState;
+  private batch = -1;
   private pendingEvents: AnyEvent<BatchExchange>[] = [];
 
   private invalidState?: InvalidAuctionStateError;
@@ -111,11 +108,11 @@ export class StreamedOrderbook {
    * @param web3 - The web3 provider to use.
    * @param options - Optional settings for tweaking the streamed orderbook.
    */
-  static async init(
+  public static async init(
     web3: Web3,
     options: Partial<OrderbookOptions> = {},
   ): Promise<StreamedOrderbook> {
-    const [contract, tx] = await batchExchangeDeployment(web3)
+    const [contract, tx] = await deployment<BatchExchange>(web3, BatchExchangeArtifact)
     const orderbook = new StreamedOrderbook(
       web3,
       contract,
@@ -131,6 +128,13 @@ export class StreamedOrderbook {
     return orderbook
   }
 
+  /**
+   * Retrieves the current open orders in the orderbook.
+   */
+  public getOpenOrders(): IndexedOrder<bigint>[] {
+    return this.state.getOrders(this.batch)
+  }
+
   /**
    * Apply all past events to the account state by querying the node for past
    * events with multiple queries to retrieve each block page at a time.
@@ -156,6 +160,7 @@ export class StreamedOrderbook {
       this.options.logger?.debug(`applying ${events.length} past events`)
       this.state.applyEvents(events)
     }
+    this.batch = await this.getBatchId(endBlock)
   }
 
   /**
@@ -181,8 +186,9 @@ export class StreamedOrderbook {
 
     const latestBlock = await this.web3.eth.getBlockNumber()
     const confirmedBlock = latestBlock - this.options.blockConfirmations
-    const confirmedEventCount = events.findIndex(ev => ev.blockNumber > confirmedBlock)
+    const batch = await this.getBatchId(confirmedBlock)
 
+    const confirmedEventCount = events.findIndex(ev => ev.blockNumber > confirmedBlock)
     const confirmedEvents = events.splice(0, confirmedEventCount)
     const pendingEvents = events
 
@@ -196,9 +202,13 @@ export class StreamedOrderbook {
         throw this.invalidState
       }
     }
+    this.batch = batch
     this.pendingEvents = pendingEvents
   }
 
+  /**
+   * Retrieves past events for the contract.
+   */
   private async getPastEvents(
     options: { fromBlock: BlockNumber; toBlock?: BlockNumber },
   ): Promise<AnyEvent<BatchExchange>[]> {
@@ -208,6 +218,23 @@ export class StreamedOrderbook {
     })
     return events as AnyEvent<BatchExchange>[]
   }
+
+  /**
+   * Retrieves the batch ID at a given block number.
+   *
+   * @remarks
+   * The batch ID is locally calculated from the block header timestamp as it is
+   * more reliable than executing an `eth_call` to calculate the batch ID on the
+   * EVM since an archive node is required for sufficiently old blocks.
+   */
+  private async getBatchId(blockNumber: BlockNumber): Promise<number> {
+    const BATCH_DURATION = 300
+
+    const block = await this.web3.eth.getBlock(blockNumber)
+    const batch = Math.floor(Number(block.timestamp) / BATCH_DURATION)
+
+    return batch
+  }
 }
 
 /**
@@ -224,15 +251,16 @@ export class InvalidAuctionStateError extends Error {
 }
 
 /**
- * Create a `BatchExchange` contract instance, returning both the web3 contract
- * object as well as the transaction receipt for the contract deployment.
+ * Get a contract deployment, returning both the web3 contract object as well as
+ * the transaction receipt for the contract deployment.
  *
  * @throws If the contract is not deployed on the network the web3 provider is
  * connected to.
  */
-async function batchExchangeDeployment(web3: Web3): Promise<[BatchExchange, TransactionReceipt]> {
-  const { abi, networks } = BatchExchangeArtifact
-
+export async function deployment<C extends Contract>(
+  web3: Web3,
+  { abi, networks }: ContractArtifact,
+): Promise<[C, TransactionReceipt]> {
   const chainId = await web3.eth.getChainId()
   const network = networks[chainId]
   if (!networks) {
@@ -242,5 +270,5 @@ async function batchExchangeDeployment(web3: Web3): Promise<[BatchExchange, Tran
   const tx = await web3.eth.getTransactionReceipt(network.transactionHash)
   const contract = new web3.eth.Contract(abi as AbiItem[], network.address)
 
-  return [contract, tx]
+  return [contract as C, tx]
 }

src/streamed/state.ts

@@ -1,6 +1,6 @@
 import assert from "assert"
 import { EventData } from "web3-eth-contract"
-import { BatchExchange } from "../.."
+import { BatchExchange, IndexedOrder } from ".."
 import { OrderbookOptions } from "."
 import { AnyEvent, Event } from "./events"
 
@@ -151,6 +151,52 @@ export class AuctionState {
     }
   }
 
+  /**
+   * Gets the current auction state in the standard order list format.
+   *
+   * @param batch - The batch to get the orders for.
+   */
+  public getOrders(batch: number): IndexedOrder<bigint>[] {
+    let orders: IndexedOrder<bigint>[] = []
+    for (const [user, account] of this.accounts.entries()) {
+      orders = orders.concat(account.orders
+        .map((order, orderId) => ({
+          ...order,
+          user,
+          sellTokenBalance: this.getEffectiveBalance(batch, user, order.sellToken),
+          orderId,
+          validUntil: order.validUntil ?? 0,
+        }))
+        .filter(order => order.validFrom <= batch && batch <= order.validUntil)
+      )
+    }
+
+    return orders
+  }
+
+  /**
+   * Retrieves a users effective balance for the specified token at a given
+   * batch.
+   *
+   * @param batch - The batch to get the balance for.
+   * @param user - The user account to retrieve the balance for.
+   * @param token - The token ID or address to retrieve the balance for.
+   */
+  private getEffectiveBalance(batch: number, user: Address, token: Address | TokenId): bigint {
+    const tokenAddr = this.tokenAddr(token)
+    const account = this.accounts.get(user)
+    if (account === undefined) {
+      return BigInt(0)
+    }
+
+    const balance = account.balances.get(tokenAddr) ?? BigInt(0)
+    const withdrawal = account.pendingWithdrawals.get(tokenAddr)
+    const withdrawalAmount = withdrawal && withdrawal.batchId <= batch ?
+      withdrawal.amount : BigInt(0)
+
+    return balance > withdrawalAmount ? balance - withdrawalAmount : BigInt(0)
+  }
+
   /**
    * Retrieves block number the account state is accepting events for.
    */
@@ -165,7 +211,6 @@ export class AuctionState {
    * This method expects that once events have been applied up until a certain
    * block number, then no new events for that block number (or an earlier block
    * number) are applied.
-   *
    */
   public applyEvents(events: AnyEvent<BatchExchange>[]): void {
     if (this.options.strict) {