Merge pull request #77 from Phala-Network/p2p

P2p

package.json

@@ -8,7 +8,7 @@
     "@chainsafe/libp2p-noise": "^5.0.0",
     "@noble/ed25519": "^1.3.3",
     "@pcan/leveldb-viewer": "^0.1.3",
-    "@phala/runtime-bridge-walkie": "^1.2.2",
+    "@phala/runtime-bridge-walkie": "^1.2.4",
     "@phala/typedefs": "^0.2.30",
     "@polkadot/api": "^7.1.1",
     "@polkadot/apps-config": "^0.100.1",
@@ -17,6 +17,7 @@
     "@polkadot/types": "^7.1.1",
     "@polkadot/util": "^8.2.2",
     "@polkadot/util-crypto": "^8.2.2",
+    "@types/duplexify": "^3.6.1",
     "agentkeepalive": "^4.1.4",
     "bee-queue": "^1.4.0",
     "bignumber.js": "^9.0.2",
@@ -51,7 +52,7 @@
     "lodash": "^4.17.21",
     "lru-cache": "^6.0.0",
     "multiaddr": "^10.0.1",
-    "multileveldown": "github:Level/multileveldown",
+    "multileveldown": "^5.0.1",
     "node-fetch": "^3.1.0",
     "p-queue": "6",
     "peer-id": "^0.16.0",
@@ -86,7 +87,8 @@
     "@types/bl": "^5.0.2",
     "@types/debug": "^4.1.7",
     "@types/ioredis": "^4.28.5",
-    "@types/levelup": "^4.3.3",
+    "@types/level-errors": "^3.0.0",
+    "@types/levelup": "^5.1.0",
     "@types/lru-cache": "^5.1.1",
     "@types/node": "^17.0.8",
     "@types/promise-retry": "^1.1.3",

src/data_provider/blob.js

@@ -198,6 +198,8 @@ export const walkWindow = async (windowId = 0, lastWindow = null) => {
   if (currentWindow) {
     parentStartBlock = currentWindow.parentStartBlock
     paraStartBlock = currentWindow.paraStartBlock
+    send('setParentCommittedHeight', parentStartBlock - 1)
+    send('setParaCommittedHeight', paraStartBlock - 1)
   } else {
     if (windowId === 0) {
       const paraId = process.env.PHALA_PARA_ID

src/data_provider/fetch_block.ts

@@ -1,4 +1,3 @@
-import { DB_BLOCK, setupDb } from './io/db'
 import {
   getLastCommittedParaBlock,
   getLastCommittedParentBlock,
@@ -11,18 +10,16 @@ import {
 } from '../utils/api'
 import { processGenesis, walkParaBlock, walkParentBlock } from './block'
 import { send } from './ipc'
-import PQueue from 'p-queue'
+import { setupDb } from './io/db'
 import env from '../utils/env'
 import logger from '../utils/logger'
 import wait from '../utils/wait'
 import type { BlockHash } from '@polkadot/types/interfaces'
 import type { U32 } from '@polkadot/types'
 import type { prb } from '@phala/runtime-bridge-walkie'
 
-const FETCH_PARENT_QUEUE_CONCURRENT = parseInt(env.parallelParentBlocks) || 15
-
 const start = async () => {
-  await setupDb(DB_BLOCK)
+  await setupDb()
   await setupParentApi(env.parentChainEndpoint)
   await setupPhalaApi(env.chainEndpoint)
 

src/data_provider/index.ts

@@ -1,14 +1,14 @@
-import { DB_BLOCK, setupDb } from './io/db'
 import { MIN_SYNCHED_DISTANCE } from '../utils/constants'
 import { fork } from './ipc'
 import { phalaApi, setupParentApi, setupPhalaApi } from '../utils/api'
 import { prb } from '@phala/runtime-bridge-walkie'
 import { processGenesis } from './block'
+import { setupDb } from './io/db'
 import { setupInternalPtp } from './ptp_int'
-import env from '../utils/env'
+import env, { dataProviderLocalServerPort } from '../utils/env'
 
 const start = async () => {
-  await setupDb(DB_BLOCK)
+  await setupDb()
   await setupParentApi(env.parentChainEndpoint)
   await setupPhalaApi(env.chainEndpoint)
 
@@ -33,6 +33,7 @@ const start = async () => {
     paraFetchedHeight: -1,
     paraProcessedHeight: -1,
     paraCommittedHeight: -1,
+    blobServerPort: dataProviderLocalServerPort,
   }
   await setupInternalPtp(genesisHash, info)
 

src/data_provider/io/block.js

@@ -1,5 +1,5 @@
-import { DB_BLOCK, NOT_FOUND_ERROR, getDb, getKeyExistence } from './db'
 import { DB_PB_TO_OBJECT_OPTIONS, pbToObject } from './db_encoding'
+import { NOT_FOUND_ERROR, getDb, getKeyExistence } from './db'
 import { prb } from '@phala/runtime-bridge-walkie'
 import logger from '../../utils/logger'
 import promiseRetry from 'promise-retry'
@@ -31,20 +31,20 @@ export const encodeBlockScale = (block, shouldCopy = false) => {
 }
 
 export const getParaBlockExistence = async (number) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   return getKeyExistence(db, `para:${number}:written`)
 }
 
 export const setParaBlock = async (number, block) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const blockPb = ParaBlock.create(block)
-  await db.set(`para:${number}:pb`, ParaBlock.encode(blockPb).finish())
-  await db.set(`para:${number}:written`, Buffer.from([1]))
+  await db.setBuffer(`para:${number}:pb`, ParaBlock.encode(blockPb).finish())
+  await db.setBuffer(`para:${number}:written`, Buffer.from([1]))
   return pbToObject(blockPb, DB_PB_TO_OBJECT_OPTIONS)
 }
 
 export const getParaBlock = async (number) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const buffer = await db.getBuffer(`para:${number}:pb`)
   if (!buffer) {
     return buffer
@@ -88,20 +88,23 @@ export const waitForParaBlock = (blockNumber) =>
   )
 
 export const getParentBlockExistance = async (number) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   return getKeyExistence(db, `parent:${number}:written`)
 }
 
 export const setParentBlock = async (number, block) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const blockPb = ParentBlock.create(block)
-  await db.set(`parent:${number}:pb`, ParentBlock.encode(blockPb).finish())
-  await db.set(`parent:${number}:written`, Buffer.from([1]))
+  await db.setBuffer(
+    `parent:${number}:pb`,
+    ParentBlock.encode(blockPb).finish()
+  )
+  await db.setBuffer(`parent:${number}:written`, Buffer.from([1]))
   return pbToObject(blockPb, DB_PB_TO_OBJECT_OPTIONS)
 }
 
 export const getParentBlock = async (number) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const buffer = await db.getBuffer(`parent:${number}:pb`)
   if (!buffer) {
     return buffer
@@ -145,15 +148,18 @@ export const waitForParentBlock = (blockNumber) =>
   )
 
 export const setGenesis = async (genesis) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const pb = Genesis.create(genesis)
-  await db.set(`genesis:${genesis.paraId}:pb`, Genesis.encode(pb).finish())
-  await db.set(`genesis:${genesis.paraId}:written`, Buffer.from([1]))
+  await db.setBuffer(
+    `genesis:${genesis.paraId}:pb`,
+    Genesis.encode(pb).finish()
+  )
+  await db.setBuffer(`genesis:${genesis.paraId}:written`, Buffer.from([1]))
   return pbToObject(pb, DB_PB_TO_OBJECT_OPTIONS)
 }
 
 export const getGenesis = async (paraId) => {
-  const db = await getDb(DB_BLOCK)
+  const db = await getDb()
   const buffer = await db.getBuffer(`genesis:${paraId}:pb`)
   if (!buffer) {
     return null