txpool refactor

e2e/package.json

@@ -25,6 +25,7 @@
 		"eslint-plugin-import": "^2.27.5",
 		"eslint-plugin-sort-imports-es6-autofix": "^0.6.0",
 		"prettier": "^2.8.4",
+		"typescript": "^4.9.5",
 		"vitest": "^0.29.2"
 	}
 }

e2e/src/author.test.ts

@@ -78,7 +78,7 @@ describe('author rpc', () => {
         invalid.resolve(result.status.toString())
       }
       if (result.status.isFinalized) {
-        finalized.resolve(null)
+        finalized.resolve()
       }
     }
 

e2e/src/block.test.ts

@@ -2,56 +2,18 @@ import { describe, expect, it } from 'vitest'
 
 import { chain, delay, dev, setupApi } from './helper'
 
-setupApi({ endpoint: 'wss://rpc.polkadot.io' })
-
 describe('block', () => {
-  it('upcoming block works', async () => {
-    const blockNumber = chain.head.number
-
-    setTimeout(() => {
-      dev.newBlock()
-    }, 1000)
-    {
-      const next = await chain.upcomingBlock()
-      expect(next.number).toEqual(blockNumber + 1)
-    }
-
-    setTimeout(() => {
-      dev.newBlock()
-    }, 1000)
-    {
-      const next = await chain.upcomingBlock()
-      expect(next.number).toEqual(blockNumber + 2)
-    }
-
-    setTimeout(() => {
-      dev.newBlock({ count: 3 })
-    }, 1000)
-    {
-      const next = await chain.upcomingBlock({ skipCount: 2 })
-      expect(next.number).toEqual(blockNumber + 5)
-    }
+  setupApi({ endpoint: 'wss://rpc.polkadot.io' })
 
-    setTimeout(() => {
-      dev.newBlock()
-    }, 1000)
-    {
-      // no block is built within 1 sec
-      await expect(chain.upcomingBlock({ timeout: 1_000 })).rejects.toThrowError('Timeout has occurred')
-
-      const next = await chain.upcomingBlock({ timeout: 10_000 })
-      expect(next.number).toEqual(blockNumber + 6)
-    }
-
-    setTimeout(() => {
-      dev.newBlock()
-    }, 1000)
-    {
-      // second block is never built
-      await expect(chain.upcomingBlock({ skipCount: 1, timeout: 10_000 })).rejects.toThrowError('Timeout has occurred')
-      expect(chain.head.number).toEqual(blockNumber + 7)
-    }
-
-    await delay(1000)
+  it('upcoming block works', async () => {
+    expect(await chain.upcomingBlocks()).toEqual(0)
+    dev.newBlock()
+    await delay(10)
+    expect(await chain.upcomingBlocks()).toEqual(1)
+
+    dev.newBlock()
+    dev.newBlock()
+    await delay(10)
+    expect(await chain.upcomingBlocks()).toEqual(2)
   })
 })

e2e/src/helper.ts

@@ -18,6 +18,7 @@ import {
 } from '@acala-network/chopsticks/src/blockchain/inherent'
 import { StorageValues } from '@acala-network/chopsticks/src/utils/set-storage'
 import { createServer } from '@acala-network/chopsticks/src/server'
+import { defer } from '@acala-network/chopsticks/src/utils'
 import { handler } from '@acala-network/chopsticks/src/rpc'
 
 export type SetupOption = {
@@ -100,13 +101,13 @@ export const setupAll = async ({
         api: apiPromise,
         async teardown() {
           await apiPromise.disconnect()
-          await new Promise((resolve) => setTimeout(resolve, 1000))
+          await delay(100)
           await close()
         },
       }
     },
     async teardownAll() {
-      await new Promise((resolve) => setTimeout(resolve, 1000))
+      await delay(100)
       await api.disconnect()
     },
   }
@@ -167,15 +168,6 @@ export const dev = {
   },
 }
 
-export function defer<T>() {
-  const deferred = {} as { resolve: (value: any) => void; reject: (reason: any) => void; promise: Promise<T> }
-  deferred.promise = new Promise((resolve, reject) => {
-    deferred.resolve = resolve
-    deferred.reject = reject
-  })
-  return deferred
-}
-
 export const mockCallback = () => {
   let next = defer()
   const callback = vi.fn((...args) => {
@@ -213,3 +205,5 @@ export const testingPairs = (ss58Format?: number) => {
     keyring,
   }
 }
+
+export { defer }

e2e/src/xcm.test.ts

@@ -1,4 +1,4 @@
-import { afterAll, describe, it } from 'vitest'
+import { afterAll, describe, expect, it } from 'vitest'
 
 import { DownwardMessage, HorizontalMessage } from '@acala-network/chopsticks/src/blockchain/txpool'
 import { connectDownward } from '@acala-network/chopsticks/src/xcm/downward'
@@ -91,7 +91,7 @@ describe('XCM', async () => {
       .signAndSend(alice)
 
     await polkadot.chain.newBlock()
-    await acala.chain.upcomingBlock()
+    expect(await acala.chain.upcomingBlocks()).toBe(1)
     await matchSnapshot(polkadot.api.query.system.events())
     await matchSnapshot(acala.api.query.system.events())
 
@@ -146,7 +146,7 @@ describe('XCM', async () => {
       .signAndSend(alice)
 
     await acala.chain.newBlock()
-    await polkadot.chain.upcomingBlock()
+    expect(await polkadot.chain.upcomingBlocks()).toBe(1)
 
     await matchSnapshot(acala.api.query.tokens.accounts(alice.address, { token: 'DOT' }))
     await matchSnapshot(polkadot.api.query.system.account(alice.address))

executor/pkg/package.json

@@ -14,4 +14,4 @@
   ],
   "main": "chopsticks_executor.js",
   "types": "chopsticks_executor.d.ts"
-}
+}

packages/chopsticks/src/blockchain/block-builder.ts

@@ -135,6 +135,7 @@ export const buildBlock = async (
   head: Block,
   inherents: HexString[],
   extrinsics: HexString[],
+  ump: Record<number, HexString[]>,
   onApplyExtrinsicError: (extrinsic: HexString, error: TransactionValidityError) => void
 ): Promise<[Block, HexString[]]> => {
   const registry = await head.registry
@@ -153,17 +154,45 @@ export const buildBlock = async (
   const pendingExtrinsics: HexString[] = []
   const includedExtrinsic: HexString[] = []
 
+  // apply ump via storage override hack
+  if (Object.keys(ump).length > 0) {
+    const meta = await head.meta
+    const layer = head.pushStorageLayer()
+    for (const [paraId, upwardMessages] of Object.entries(ump)) {
+      const queueSize = meta.registry.createType('(u32, u32)', [
+        upwardMessages.length,
+        upwardMessages.map((x) => x.length).reduce((s, i) => s + i, 0),
+      ])
+
+      const messages = meta.registry.createType('Vec<Bytes>', upwardMessages)
+
+      // TODO: make sure we append instead of replace
+      layer.setAll([
+        [compactHex(meta.query.ump.relayDispatchQueues(paraId)), messages.toHex()],
+        [compactHex(meta.query.ump.relayDispatchQueueSize(paraId)), queueSize.toHex()],
+      ])
+    }
+
+    logger.trace(
+      {
+        number: newBlock.number,
+        tempHash: newBlock.hash,
+        ump,
+      },
+      'Upward messages'
+    )
+
+    const needsDispatch = meta.registry.createType('Vec<u32>', Object.keys(ump))
+    layer.set(compactHex(meta.query.ump.needsDispatch()), needsDispatch.toHex())
+  }
+
   // apply extrinsics
   for (const extrinsic of extrinsics) {
     try {
       const { result, storageDiff } = await newBlock.call('BlockBuilder_apply_extrinsic', [extrinsic])
       const outcome = registry.createType<ApplyExtrinsicResult>('ApplyExtrinsicResult', result)
       if (outcome.isErr) {
-        if (outcome.asErr.isInvalid && outcome.asErr.asInvalid.isFuture) {
-          pendingExtrinsics.push(extrinsic)
-        } else {
-          onApplyExtrinsicError(extrinsic, outcome.asErr)
-        }
+        onApplyExtrinsicError(extrinsic, outcome.asErr)
         continue
       }
       newBlock.pushStorageLayer().setAll(storageDiff)

packages/chopsticks/src/blockchain/index.ts

@@ -8,14 +8,7 @@ import type { TransactionValidity } from '@polkadot/types/interfaces/txqueue'
 
 import { Api } from '../api'
 import { Block } from './block'
-import {
-  BuildBlockMode,
-  BuildBlockParams,
-  DownwardMessage,
-  HorizontalMessage,
-  TxPool,
-  UpcomingBlockParams,
-} from './txpool'
+import { BuildBlockMode, BuildBlockParams, DownwardMessage, HorizontalMessage, TxPool } from './txpool'
 import { HeadState } from './head-state'
 import { InherentProvider } from './inherent'
 import { StorageValue } from './storage-layer'
@@ -170,13 +163,31 @@ export class Blockchain {
     throw validity.asErr
   }
 
+  submitUpwardMessages(id: number, ump: HexString[]) {
+    this.#txpool.submitUpwardMessages(id, ump)
+
+    logger.debug({ id, ump }, 'submitUpwardMessages')
+  }
+
+  submitDownwardMessages(dmp: DownwardMessage[]) {
+    this.#txpool.submitDownwardMessages(dmp)
+
+    logger.debug({ dmp }, 'submitDownwardMessages')
+  }
+
+  submitHorizontalMessages(id: number, hrmp: HorizontalMessage[]) {
+    this.#txpool.submitHorizontalMessages(id, hrmp)
+
+    logger.debug({ id, hrmp }, 'submitHorizontalMessages')
+  }
+
   async newBlock(params?: BuildBlockParams): Promise<Block> {
     await this.#txpool.buildBlock(params)
     return this.#head
   }
 
-  async upcomingBlock(params?: UpcomingBlockParams) {
-    return this.#txpool.upcomingBlock(params)
+  async upcomingBlocks() {
+    return this.#txpool.upcomingBlocks()
   }
 
   async dryRunExtrinsic(

packages/chopsticks/src/blockchain/inherent/index.ts

@@ -10,7 +10,7 @@ export { SetBabeRandomness } from './parachain/babe-randomness'
 export { SetNimbusAuthorInherent } from './parachain/nimbus-author-inherent'
 
 export interface CreateInherents {
-  createInherents(parent: Block, params?: BuildBlockParams['inherent']): Promise<HexString[]>
+  createInherents(parent: Block, params: BuildBlockParams): Promise<HexString[]>
 }
 
 export type InherentProvider = CreateInherents
@@ -33,7 +33,7 @@ export class InherentProviders implements InherentProvider {
     this.#providers = providers
   }
 
-  async createInherents(parent: Block, params?: BuildBlockParams['inherent']): Promise<HexString[]> {
+  async createInherents(parent: Block, params: BuildBlockParams): Promise<HexString[]> {
     const base = await this.#base.createInherents(parent, params)
     const extra = await Promise.all(this.#providers.map((provider) => provider.createInherents(parent, params)))
     return [...base, ...extra.flat()]

packages/chopsticks/src/blockchain/inherent/para-enter.ts

@@ -6,7 +6,7 @@ import { BuildBlockParams } from '../txpool'
 import { CreateInherents } from '.'
 
 export class ParaInherentEnter implements CreateInherents {
-  async createInherents(parent: Block, _params?: BuildBlockParams['inherent']): Promise<HexString[]> {
+  async createInherents(parent: Block, _params: BuildBlockParams): Promise<HexString[]> {
     const meta = await parent.meta
     if (!meta.tx.paraInherent?.enter) {
       return []

packages/chopsticks/src/blockchain/inherent/parachain/babe-randomness.ts

@@ -7,7 +7,7 @@ import { CreateInherents } from '..'
 
 // Support for Moonbeam pallet-randomness mandatory inherent
 export class SetBabeRandomness implements CreateInherents {
-  async createInherents(parent: Block, _params?: BuildBlockParams['inherent']): Promise<HexString[]> {
+  async createInherents(parent: Block, _params: BuildBlockParams): Promise<HexString[]> {
     const meta = await parent.meta
     if (!meta.tx.randomness?.setBabeRandomnessResults) {
       return []

packages/chopsticks/src/blockchain/inherent/parachain/nimbus-author-inherent.ts

@@ -7,7 +7,7 @@ import { CreateInherents } from '..'
 
 // Support for Nimbus Author Inherent
 export class SetNimbusAuthorInherent implements CreateInherents {
-  async createInherents(parent: Block, _params?: BuildBlockParams['inherent']): Promise<HexString[]> {
+  async createInherents(parent: Block, _params: BuildBlockParams): Promise<HexString[]> {
     const meta = await parent.meta
     if (!meta.tx.authorInherent?.kickOffAuthorshipValidation) {
       return []

packages/chopsticks/src/blockchain/inherent/parachain/validation-data.ts

@@ -58,7 +58,7 @@ export type ValidationData = {
 }
 
 export class SetValidationData implements CreateInherents {
-  async createInherents(parent: Block, params?: BuildBlockParams['inherent']): Promise<HexString[]> {
+  async createInherents(parent: Block, params: BuildBlockParams): Promise<HexString[]> {
     const meta = await parent.meta
     if (!meta.tx.parachainSystem?.setValidationData) {
       return []
@@ -109,19 +109,19 @@ export class SetValidationData implements CreateInherents {
       // inject downward messages
       let dmqMqcHeadHash = decoded[dmqMqcHeadKey]
       if (dmqMqcHeadHash) {
-        for (const { msg, sentAt } of params?.downwardMessages || []) {
+        for (const { data, sentAt } of params.downwardMessages || []) {
           // calculate new hash
           dmqMqcHeadHash = blake2AsHex(
             u8aConcat(
               meta.registry.createType('Hash', dmqMqcHeadHash).toU8a(),
               meta.registry.createType('BlockNumber', sentAt).toU8a(),
-              blake2AsU8a(meta.registry.createType('Bytes', msg).toU8a(), 256)
+              blake2AsU8a(meta.registry.createType('Bytes', data).toU8a(), 256)
             ),
             256
           )
 
           downwardMessages.push({
-            msg,
+            data,
             sentAt,
           })
         }