Mock ingest transactions & produce blocks

packages/core/package.json

@@ -29,6 +29,7 @@
   },
   "dependencies": {
     "abstract-leveldown": "^6.0.3",
+    "async-mutex": "^0.1.3",
     "axios": "^0.19.0",
     "bn.js": "^4.11.8",
     "body-parser": "^1.19.0",

packages/core/src/app/aggregator/aggregator.ts

@@ -0,0 +1,62 @@
+import BigNum = require('bn.js')
+
+import { Aggregator } from '../../types/aggregator'
+import { StateManager } from '../../types/ovm'
+import {
+  BlockTransaction,
+  BlockTransactionCommitment,
+  Transaction,
+  TransactionResult,
+} from '../../types/serialization'
+import { rangesSpanRange, sign } from '../utils'
+import { BlockManager } from '../../types/block-production'
+
+export class DefaultAggregator implements Aggregator {
+  private readonly publicKey: string =
+    'TODO: figure out public key storage and access'
+  private readonly privateKey: string =
+    'TODO: figure out private key storage and access'
+
+  public constructor(
+    private readonly stateManager: StateManager,
+    private readonly blockManager: BlockManager
+  ) {}
+
+  public async ingestTransaction(
+    transaction: Transaction,
+    witness: string
+  ): Promise<BlockTransactionCommitment> {
+    const blockNumber: BigNum = await this.blockManager.getNextBlockNumber()
+
+    const {
+      stateUpdate,
+      validRanges,
+    }: TransactionResult = await this.stateManager.executeTransaction(
+      transaction,
+      blockNumber,
+      witness
+    )
+
+    if (!rangesSpanRange(validRanges, transaction.range)) {
+      throw Error(
+        'Cannot ingest Transaction that is not valid across its entire range.'
+      )
+    }
+
+    await this.blockManager.addPendingStateUpdate(stateUpdate)
+
+    const blockTransaction: BlockTransaction = {
+      blockNumber,
+      transaction,
+    }
+
+    return {
+      blockTransaction,
+      witness: sign(this.privateKey, blockTransaction),
+    }
+  }
+
+  public async getPublicKey(): Promise<any> {
+    return this.publicKey
+  }
+}

packages/core/src/app/aggregator/index.ts

@@ -0,0 +1 @@
+export * from './aggregator'

packages/core/src/app/block-production/block-db.ts

@@ -0,0 +1,151 @@
+/* External Imports */
+import BigNum = require('bn.js')
+import { Mutex } from 'async-mutex'
+
+import { BaseKey, BaseRangeBucket } from '../db'
+import { BlockDB } from '../../types/block-production'
+import { KeyValueStore, RangeStore } from '../../types/db'
+import { StateUpdate } from '../../types/serialization'
+import { MAX_BIG_NUM, ONE, ZERO } from '../utils'
+import { GenericMerkleIntervalTree } from './merkle-interval-tree'
+import { deserializeStateUpdate, serializeStateUpdate } from '../serialization'
+
+const KEYS = {
+  NEXT_BLOCK: Buffer.from('nextblock'),
+  BLOCK: new BaseKey('b', ['buffer']),
+}
+
+/**
+ * Simple BlockDB implementation.
+ */
+export class DefaultBlockDB implements BlockDB {
+  private readonly blockMutex: Mutex
+
+  /**
+   * Initializes the database wrapper.
+   * @param vars the KeyValueStore to store variables in
+   * @param blocks the KeyValueStore to store Blocks in
+   */
+  constructor(
+    private readonly vars: KeyValueStore,
+    private readonly blocks: KeyValueStore
+  ) {
+    this.blockMutex = new Mutex()
+  }
+
+  /**
+   * @returns the next plasma block number.
+   */
+  public async getNextBlockNumber(): Promise<BigNum> {
+    // TODO: Cache this when it makes sense
+    const buf = await this.vars.get(KEYS.NEXT_BLOCK)
+    return !buf ? ONE : new BigNum(buf, 'be')
+  }
+
+  /**
+   * Adds a state update to the list of updates to be published in the next
+   * plasma block.
+   * @param stateUpdate State update to publish in the next block.
+   * @returns a promise that resolves once the update has been added.
+   */
+  public async addPendingStateUpdate(stateUpdate: StateUpdate): Promise<void> {
+    await this.blockMutex.runExclusive(async () => {
+      const block = await this.getNextBlockStore()
+      const start = stateUpdate.range.start
+      const end = stateUpdate.range.end
+
+      if (await block.hasDataInRange(start, end)) {
+        throw new Error(
+          'Block already contains a state update over that range.'
+        )
+      }
+
+      const value = Buffer.from(serializeStateUpdate(stateUpdate))
+      await block.put(start, end, value)
+    })
+  }
+
+  /**
+   * @returns the list of state updates waiting to be published in the next
+   * plasma block.
+   */
+  public async getPendingStateUpdates(): Promise<StateUpdate[]> {
+    const blockNumber = await this.getNextBlockNumber()
+    return this.getStateUpdates(blockNumber)
+  }
+
+  /**
+   * Computes the Merkle Interval Tree root of a given block.
+   * @param blockNumber Block to compute a root for.
+   * @returns the root of the block.
+   */
+  public async getMerkleRoot(blockNumber: BigNum): Promise<Buffer> {
+    const stateUpdates = await this.getStateUpdates(blockNumber)
+
+    const leaves = stateUpdates.map((stateUpdate) => {
+      // TODO: Actually encode this.
+      const encodedStateUpdate = serializeStateUpdate(stateUpdate)
+      return {
+        start: stateUpdate.range.start,
+        end: stateUpdate.range.end,
+        data: encodedStateUpdate,
+      }
+    })
+    const tree = new GenericMerkleIntervalTree(leaves)
+    return tree.root().hash
+  }
+
+  /**
+   * Finalizes the next plasma block so that it can be published.
+   *
+   * Note: The execution of this function is serialized internally,
+   * but to be of use, the caller will most likely want to serialize
+   * their calls to it as well.
+   */
+  public async finalizeNextBlock(): Promise<void> {
+    await this.blockMutex.runExclusive(async () => {
+      const prevBlockNumber: BigNum = await this.getNextBlockNumber()
+      const nextBlockNumber: Buffer = prevBlockNumber.add(ONE).toBuffer('be')
+
+      await this.vars.put(KEYS.NEXT_BLOCK, nextBlockNumber)
+    })
+  }
+
+  /**
+   * Opens the RangeDB for a specific block.
+   * @param blockNumber Block to open the RangeDB for.
+   * @returns the RangeDB instance for the given block.
+   */
+  private async getBlockStore(blockNumber: BigNum): Promise<RangeStore> {
+    const key = KEYS.BLOCK.encode([blockNumber.toBuffer('be')])
+    const bucket = this.blocks.bucket(key)
+    return new BaseRangeBucket(bucket.db, bucket.prefix)
+  }
+
+  /**
+   * @returns the RangeDB instance for the next block to be published.
+   *
+   * IMPORTANT: This function itself is safe from concurrency issues, but
+   * if the caller is modifying the returned RangeStore or needs to
+   * guarantee the returned next RangeStore is not stale, both the call
+   * to this function AND any subsequent reads / writes should be run with
+   * the blockMutex lock held to guarantee the expected behavior.
+   */
+  private async getNextBlockStore(): Promise<RangeStore> {
+    const blockNumber = await this.getNextBlockNumber()
+    return this.getBlockStore(blockNumber)
+  }
+
+  /**
+   * Queries all of the state updates within a given block.
+   * @param blockNumber Block to query state updates for.
+   * @returns the list of state updates for that block.
+   */
+  private async getStateUpdates(blockNumber: BigNum): Promise<StateUpdate[]> {
+    const block = await this.getBlockStore(blockNumber)
+    const values = await block.get(ZERO, MAX_BIG_NUM)
+    return values.map((value) => {
+      return deserializeStateUpdate(value.value.toString())
+    })
+  }
+}

packages/core/src/app/block-production/block-manager.ts

@@ -0,0 +1,69 @@
+import BigNum = require('bn.js')
+import { Mutex } from 'async-mutex'
+
+import {
+  BlockDB,
+  BlockManager,
+  CommitmentContract,
+} from '../../types/block-production'
+import { StateUpdate } from '../../types/serialization'
+
+/**
+ * Simple BlockManager implementation.
+ */
+export class DefaultBlockManager implements BlockManager {
+  private readonly blockSubmissionMutex: Mutex
+
+  /**
+   * Initializes the manager.
+   * @param blockdb BlockDB instance to store/query data from.
+   * @param commitmentContract Contract wrapper used to publish block roots.
+   */
+  constructor(
+    private blockdb: BlockDB,
+    private commitmentContract: CommitmentContract
+  ) {
+    this.blockSubmissionMutex = new Mutex()
+  }
+
+  /**
+   * @returns the next plasma block number.
+   */
+  public async getNextBlockNumber(): Promise<BigNum> {
+    return this.blockdb.getNextBlockNumber()
+  }
+
+  /**
+   * Adds a state update to the list of updates to be published in the next
+   * plasma block.
+   * @param stateUpdate State update to add to the next block.
+   * @returns a promise that resolves once the update has been added.
+   */
+  public async addPendingStateUpdate(stateUpdate: StateUpdate): Promise<void> {
+    await this.blockdb.addPendingStateUpdate(stateUpdate)
+  }
+
+  /**
+   * @returns the state updates to be published in the next block.
+   */
+  public async getPendingStateUpdates(): Promise<StateUpdate[]> {
+    return this.blockdb.getPendingStateUpdates()
+  }
+
+  /**
+   * Finalizes the next block and submits the block root to Ethereum.
+   * @returns a promise that resolves once the block has been published.
+   */
+  public async submitNextBlock(): Promise<void> {
+    await this.blockSubmissionMutex.runExclusive(async () => {
+      // Don't submit the block if there are no StateUpdates
+      if ((await this.getPendingStateUpdates()).length === 0) {
+        return
+      }
+      const blockNumber = await this.getNextBlockNumber()
+      await this.blockdb.finalizeNextBlock()
+      const root = await this.blockdb.getMerkleRoot(blockNumber)
+      await this.commitmentContract.submitBlock(root)
+    })
+  }
+}

packages/core/src/app/block-production/index.ts

@@ -1,3 +1,5 @@
+export * from './block-db'
+export * from './block-manager'
 export * from './merkle-interval-tree'
 export * from './state-interval-tree'
 export * from './plasma-block-tree'

packages/core/src/app/block-production/plasma-block-tree.ts

@@ -42,7 +42,7 @@ export class PlasmaBlock extends GenericMerkleIntervalTree
 
   /**
    * Returns a double inclusion proof which demonstrates the existence of a state update within the plasma block.
-   * @param stateUpdatePosition index of the state udpate in the state subtree of the block.
+   * @param stateUpdatePosition index of the state update in the state subtree of the block.
    * @param assetIdPosition index of the assetId in the top-level asset id of the block
    */
   public getStateUpdateInclusionProof(

packages/core/src/app/db/db.ts

@@ -4,7 +4,11 @@
  */
 
 /* External Imports */
-import { AbstractOpenOptions, AbstractLevelDOWN } from 'abstract-leveldown'
+import {
+  AbstractOpenOptions,
+  AbstractLevelDOWN,
+  AbstractChainedBatch,
+} from 'abstract-leveldown'
 
 /* Internal Imports */
 import {
@@ -16,12 +20,18 @@ import {
   Iterator,
   Bucket,
   RangeBucket,
+  KeyValueStore,
+  PutBatch,
+  PUT_BATCH_TYPE,
+  DEL_BATCH_TYPE,
 } from '../../types'
 import { BaseIterator } from './iterator'
 import { BaseBucket } from './bucket'
 import { BaseRangeBucket } from './range-bucket'
 import { bufferUtils } from '../../app'
 
+export const DEFAULT_PREFIX_LENGTH = 3
+
 /**
  * Checks if an error is a NotFoundError.
  * @param err Error to check.
@@ -45,7 +55,7 @@ const isNotFound = (err: any): boolean => {
 export class BaseDB implements DB {
   constructor(
     readonly db: AbstractLevelDOWN,
-    readonly prefixLength: number = 3
+    readonly prefixLength: number = DEFAULT_PREFIX_LENGTH
   ) {}
 
   /**

packages/core/src/app/db/range-bucket.ts

@@ -18,6 +18,7 @@ import {
   RangeBucket,
   RangeEntry,
   Endianness,
+  PUT_BATCH_TYPE,
 } from '../../types'
 import { bufferUtils, intersects, BaseDB } from '../../app'
 
@@ -194,15 +195,15 @@ export class BaseRangeBucket implements RangeBucket {
     if (ranges.length > 0 && start.gt(ranges[0].start)) {
       // Reduce the first affected range's end position. Eg: ##### becomes ###$$
       batchOps.push({
-        type: 'put',
+        type: PUT_BATCH_TYPE,
         key: this.bnToKey(start),
         value: this.addStartToValue(ranges[0].start, ranges[0].value),
       })
     }
     // If the end position less than the last range's end...
     if (ranges.length > 0 && ranges[ranges.length - 1].end.gt(end)) {
       batchOps.push({
-        type: 'put',
+        type: PUT_BATCH_TYPE,
         key: this.bnToKey(ranges[ranges.length - 1].end),
         value: this.addStartToValue(end, ranges[ranges.length - 1].value),
       })
@@ -211,7 +212,7 @@ export class BaseRangeBucket implements RangeBucket {
     // Step #3: Add our new range
     //
     batchOps.push({
-      type: 'put',
+      type: PUT_BATCH_TYPE,
       key: this.bnToKey(end),
       value: this.addStartToValue(start, value),
     })
@@ -234,6 +235,11 @@ export class BaseRangeBucket implements RangeBucket {
     return ranges
   }
 
+  public async hasDataInRange(start: BigNum, end: BigNum): Promise<boolean> {
+    // TODO: can eagerly return when true, but this is good enough or now
+    return (await this.get(start, end)).length > 0
+  }
+
   /**
    * Gets all ranges which intersect with [start,end)
    * @param start The start of the range we are getting.