Hive-interop-updates

packages/cli/src/rpc/modules/portal.ts

@@ -19,6 +19,7 @@ import {
   ContentMessageType,
   AcceptMessage,
   decodeHistoryNetworkContentKey,
+  FoundContent,
   BeaconLightClientNetwork,
   BeaconLightClientNetworkContentType,
 } from 'portalnetwork'
@@ -217,7 +218,8 @@ export class portal {
       if (this._history.routingTable.getWithPending(encodedENR.nodeId)?.value) {
         return true
       }
-      this._history.routingTable.insertOrUpdate(encodedENR, EntryStatus.Disconnected)
+      this._client.discv5.addEnr(enr)
+      this._history.routingTable.insertOrUpdate(encodedENR, EntryStatus.Connected)
       return true
     } catch {
       return false
@@ -305,18 +307,13 @@ export class portal {
   }
   async historyFindNodes(params: [string, number[]]) {
     const [enr, distances] = params
-    this.logger(`findNodes request received with these distances ${distances.toString()}`)
     const dstId = ENR.decodeTxt(enr).nodeId
+    this.logger(`findNodes request received with these distances [${distances.toString()}]`)
+    this.logger(`sending findNodes request to ${shortId(dstId)}`)
     if (!isValidId(dstId)) {
       return 'invalid node id'
     }
-    if (!this._history.routingTable.getWithPending(dstId)?.value) {
-      const pong = await this._history.sendPing(enr)
-      if (!pong) {
-        return ''
-      }
-    }
-    const res = await this._history.sendFindNodes(dstId, distances)
+    const res = await this._history.sendFindNodes(enr, distances)
     if (!res) {
       return []
     }
@@ -396,11 +393,16 @@ export class portal {
       }
     }
     const res = await this._history.sendFindContent(nodeId, fromHexString(contentKey))
+    this.logger.extend('findContent')(
+      `request returned type: ${res ? FoundContent[res.selector] : res}`,
+    )
     if (!res) {
-      return ''
+      return { enrs: [] }
     }
-    const content: Uint8Array =
-      res.selector === 1
+    const content: Uint8Array | Uint8Array[] =
+      res.selector === FoundContent.ENRS
+        ? (res.value as Uint8Array[])
+        : res.selector === FoundContent.CONTENT
         ? (res.value as Uint8Array)
         : await new Promise((resolve) => {
             const timeout = setTimeout(() => {
@@ -417,12 +419,14 @@ export class portal {
             )
           })
     this.logger.extend('findContent')(`request returned ${content.length} bytes`)
-    res.selector === 0 && this.logger.extend('findContent')('utp')
+    res.selector === FoundContent.UTP && this.logger.extend('findContent')('utp')
     this.logger.extend('findContent')(content)
-    return {
-      content: content.length > 0 ? toHexString(content) : '',
-      utpTransfer: res.selector === 0,
-    }
+    return res.selector === FoundContent.ENRS
+      ? { enrs: content }
+      : {
+          content: content.length > 0 ? toHexString(content as Uint8Array) : '',
+          utpTransfer: res.selector === FoundContent.UTP,
+        }
   }
   async historySendFindContent(params: [string, string]) {
     const [nodeId, contentKey] = params
@@ -446,12 +450,28 @@ export class portal {
   }
   async historyRecursiveFindContent(params: [string]) {
     const [contentKey] = params
+    this.logger.extend('historyRecursiveFindContent')(`request received for ${contentKey}`)
     const lookup = new ContentLookup(this._history, fromHexString(contentKey))
     const res = await lookup.startLookup()
-    if (res instanceof Uint8Array) {
-      return toHexString(res)
+    this.logger.extend('historyRecursiveFindContent')(`request returned ${JSON.stringify(res)}`)
+    if (!res) {
+      this.logger.extend('historyRecursiveFindContent')(`request returned { enrs: [] }`)
+      return { content: '0x', utpTransfer: false }
+    }
+    if ('enrs' in res) {
+      this.logger.extend('historyRecursiveFindContent')(
+        `request returned { enrs: [{${{ enrs: res.enrs.map(toHexString) }}}] }`,
+      )
+      return { enrs: res.enrs.map(toHexString) }
+    } else {
+      this.logger.extend('historyRecursiveFindContent')(
+        `request returned { content: ${toHexString(res.content)}, utpTransfer: ${res.utp} }`,
+      )
+      return {
+        content: toHexString(res.content),
+        utpTransfer: res.utp,
+      }
     }
-    return res
   }
   async historyOffer(params: [string, string, string]) {
     const [enrHex, contentKeyHex, contentValueHex] = params

packages/cli/src/rpc/modules/ultralight.ts

@@ -53,7 +53,9 @@ export class ultralight {
     const [contentKey, value] = params
 
     const type: number = parseInt(contentKey.slice(0, 4))
-    this.logger(`ultralight_addContentToDB request received for ${HistoryNetworkContentType[type]} ${contentKey}`)
+    this.logger(
+      `ultralight_addContentToDB request received for ${HistoryNetworkContentType[type]} ${contentKey}`,
+    )
     try {
       this._history.store(type, '0x' + contentKey.slice(4), fromHexString(value))
       this.logger(`${type} value for 0x${contentKey.slice(4)} added to content DB`)

packages/cli/src/rpc/validators.ts

@@ -132,12 +132,15 @@ export const validators = {
           message: `invalid argument ${index}: argument must be a number`,
         }
       }
-      if (params[index] < 0 || params[index] > 255) {
+      if (params[index] < 0 || params[index] > 256) {
         return {
           code: INVALID_PARAMS,
-          message: 'invalid argument ${index}: distance must be between 0 and 255.',
+          message: `invalid argument ${index}: argument must be a number`,
         }
       }
+      if (params[index] === 256) {
+        params[index] = 0
+      }
     }
   },
   /**

packages/portalnetwork/src/client/client.ts

@@ -10,7 +10,7 @@ import { ITalkReqMessage, ITalkRespMessage } from '@chainsafe/discv5/message'
 import { EventEmitter } from 'events'
 import debug, { Debugger } from 'debug'
 import { fromHexString, toHexString } from '@chainsafe/ssz'
-import { BeaconLightClientNetwork, StateProtocol } from '../subprotocols/index.js'
+import { BeaconLightClientNetwork, StateProtocol, ProtocolId } from '../subprotocols/index.js'
 import {
   PortalNetworkEventEmitter,
   PortalNetworkMetrics,
@@ -21,7 +21,7 @@ import type { PeerId, Secp256k1PeerId } from '@libp2p/interface-peer-id'
 import { createSecp256k1PeerId } from '@libp2p/peer-id-factory'
 import { INodeAddress } from '@chainsafe/discv5/lib/session/nodeInfo.js'
 import { PortalNetworkUTP } from '../wire/utp/PortalNetworkUtp/index.js'
-import { ProtocolId } from '../types.js'
+
 import { BaseProtocol } from '../subprotocols/protocol.js'
 import { HistoryProtocol } from '../subprotocols/history/history.js'
 import { Multiaddr, multiaddr } from '@multiformats/multiaddr'
@@ -305,8 +305,8 @@ export class PortalNetwork extends (EventEmitter as { new (): PortalNetworkEvent
           value: JSON.stringify(peers),
         },
       ])
-    } catch (err) {
-      console.log('error', err)
+    } catch (err: any) {
+      this.logger.log('error: ', err.message)
     }
   }
 

packages/portalnetwork/src/client/provider.ts

@@ -1,6 +1,6 @@
 import { ethers } from 'ethers'
 import { addRLPSerializedBlock, HistoryProtocol } from '../subprotocols/index.js'
-import { ProtocolId } from '../types.js'
+import { ProtocolId } from '../subprotocols/types.js'
 import { toHexString } from '../util/discv5.js'
 import {
   ethJsBlockToEthersBlock,

packages/portalnetwork/src/index.ts

@@ -5,4 +5,3 @@ export * from './util/index.js'
 export * from './transports/index.js'
 export * from './subprotocols/index.js'
 export * from './client/index.js'
-export * from './types.js'

packages/portalnetwork/src/subprotocols/beacon/beacon.ts

@@ -1,7 +1,6 @@
 import { Debugger } from 'debug'
 import { BaseProtocol } from '../protocol.js'
-import { FoundContent } from '../types.js'
-import { ProtocolId } from '../../types.js'
+import { ProtocolId } from '../types.js'
 import { PortalNetwork } from '../../client/client.js'
 import debug from 'debug'
 import { Union } from '@chainsafe/ssz/lib/interface.js'
@@ -20,7 +19,7 @@ import {
   PortalWireMessageType,
 } from '../../wire/types.js'
 import { bytesToInt } from '@ethereumjs/util'
-import { RequestCode } from '../../wire/index.js'
+import { RequestCode, FoundContent } from '../../wire/index.js'
 import { ssz } from '@lodestar/types'
 export class BeaconLightClientNetwork extends BaseProtocol {
   protocolId: ProtocolId.BeaconLightClientNetwork
@@ -77,6 +76,7 @@ export class BeaconLightClientNetwork extends BaseProtocol {
     if (res.length === 0) {
       return undefined
     }
+
     try {
       if (bytesToInt(res.subarray(0, 1)) === MessageCodes.CONTENT) {
         this.metrics?.contentMessagesReceived.inc()

packages/portalnetwork/src/subprotocols/contentLookup.ts

@@ -11,6 +11,14 @@ type lookupPeer = {
   hasContent?: boolean
 }
 
+export type ContentLookupResponse =
+  | {
+      content: Uint8Array
+      utp: boolean
+    }
+  | { enrs: Uint8Array[] }
+  | undefined
+
 export class ContentLookup {
   private protocol: BaseProtocol
   private lookupPeers: lookupPeer[]
@@ -33,13 +41,13 @@ export class ContentLookup {
    * Queries the 5 nearest nodes in the history network routing table and recursively
    * requests peers closer to the content until either the content is found or there are no more peers to query
    */
-  public startLookup = async (): Promise<Uint8Array | Uint8Array[] | undefined> => {
+  public startLookup = async (): Promise<ContentLookupResponse> => {
     // Don't support content lookups for protocols that don't implement it (i.e. Canonical Indices)
     if (!this.protocol.sendFindContent) return
     this.protocol.metrics?.totalContentLookups.inc()
     try {
       const res = await this.protocol.get(this.protocol.protocolId, toHexString(this.contentKey))
-      return fromHexString(res)
+      return { content: fromHexString(res), utp: false }
     } catch (err: any) {
       this.logger(`content key not in db ${err.message}`)
     }
@@ -77,8 +85,17 @@ export class ContentLookup {
               contentType: HistoryNetworkContentType,
               content: string,
             ) => {
-              this.protocol.removeListener('ContentAdded', utpDecoder)
-              resolve(fromHexString(content))
+              this.logger(
+                `this.contentKey: ${contentType} +  ${toHexString(this.contentKey.slice(1))}`,
+              )
+              this.logger(`contentType: ${contentType} contentKey: ${contentKey}, .`)
+              if (
+                this.contentKey[0] === contentType &&
+                contentKey === toHexString(this.contentKey.slice(1))
+              ) {
+                this.protocol.removeListener('ContentAdded', utpDecoder)
+                resolve({ content: fromHexString(content), utp: true })
+              }
             }
             this.protocol.on('ContentAdded', utpDecoder)
           })
@@ -99,7 +116,7 @@ export class ContentLookup {
               this.protocol.sendOffer(peer, [this.contentKey])
             }
           }
-          return res.value
+          return { content: res.value as Uint8Array, utp: false }
         }
         case 2: {
           // findContent request returned ENRs of nodes closer to content

packages/portalnetwork/src/subprotocols/history/eth_module.ts

@@ -10,7 +10,7 @@ import {
   epochRootByBlocknumber,
   BlockHeaderWithProof,
 } from './index.js'
-import { ContentLookup } from '../index.js'
+import { ContentLookup, ContentLookupResponse } from '../index.js'
 
 export class ETH {
   protocol: HistoryProtocol
@@ -28,15 +28,17 @@ export class ETH {
     const bodyContentKey = includeTransactions
       ? fromHexString(getContentKey(HistoryNetworkContentType.BlockBody, fromHexString(blockHash)))
       : undefined
+    let lookupResponse: ContentLookupResponse
     let header: any
     let body: any
     let block
     try {
       let lookup = new ContentLookup(this.protocol, headerContentKey)
-      header = await lookup.startLookup()
-      if (!(header instanceof Uint8Array)) {
+      lookupResponse = await lookup.startLookup()
+      if (!lookupResponse || !('content' in lookupResponse)) {
         return undefined
       } else {
+        header = lookupResponse.content
         header = BlockHeaderWithProof.deserialize(header as Uint8Array).header
       }
       if (!includeTransactions) {
@@ -50,11 +52,12 @@ export class ETH {
         return block
       } else {
         lookup = new ContentLookup(this.protocol, bodyContentKey!)
-        body = await lookup.startLookup()
-        try {
-          block = reassembleBlock(header, body)
-        } catch {
+        lookupResponse = await lookup.startLookup()
+        if (!lookupResponse || !('content' in lookupResponse)) {
           block = reassembleBlock(header)
+        } else {
+          body = lookupResponse.content
+          block = reassembleBlock(header, body)
         }
       }
     } catch {}
@@ -71,11 +74,11 @@ export class ETH {
     const epoch_lookup = new ContentLookup(this.protocol, fromHexString(lookupKey))
     const result = await epoch_lookup.startLookup()
 
-    if (result instanceof Uint8Array) {
+    if (result && 'content' in result) {
       this.protocol.logger.extend(`ETH_GETBLOCKBYNUMBER`)(
         `Found EpochAccumulator with header record for block ${blockNumber}`,
       )
-      const epoch = EpochAccumulator.deserialize(result)
+      const epoch = EpochAccumulator.deserialize(result.content)
       blockHash = toHexString(epoch[Number(blockNumber) % 8192].blockHash)
 
       const block = await this.getBlockByHash(blockHash, includeTransactions)

packages/portalnetwork/src/subprotocols/history/history.ts

@@ -15,7 +15,7 @@ import {
   PortalNetwork,
   FoundContent,
 } from '../../index.js'
-import { ProtocolId } from '../../types.js'
+import { ProtocolId } from '../types.js'
 import { ETH } from './eth_module.js'
 import { GossipManager } from './gossip.js'
 import { BlockHeaderWithProof, EpochAccumulator, HistoryNetworkContentType } from './types.js'
@@ -196,7 +196,7 @@ export class HistoryProtocol extends BaseProtocol {
       const content = await this.get(this.protocolId, contentKey)
       return content
     } catch {
-      this.logger('Error retrieving content from DB')
+      this.logger(`Error retrieving ${contentKey} from DB`)
     }
   }
 
@@ -227,18 +227,21 @@ export class HistoryProtocol extends BaseProtocol {
       this.logger(`Block Header for ${shortId(hashKey)} not found locally.  Querying network...`)
       block = await this.ETH.getBlockByHash(hashKey, false)
     }
+    const bodyContentKey = getContentKey(
+      HistoryNetworkContentType.BlockBody,
+      fromHexString(hashKey),
+    )
     if (block instanceof Block) {
-      const bodyContentKey = getContentKey(
-        HistoryNetworkContentType.BlockBody,
-        fromHexString(hashKey),
-      )
       this.put(this.protocolId, bodyContentKey, toHexString(value))
       if (block.transactions.length > 0) {
         await this.saveReceipts(block)
       }
     } else {
       this.logger(`Could not verify block content`)
-      // Don't store block body where we can't assemble a valid block
+      this.logger(`Adding anyway for testing...`)
+      this.put(this.protocolId, bodyContentKey, toHexString(value))
+      // TODO: Decide what to do here.  We shouldn't be storing block bodies without a corresponding header
+      // as it's against spec
       return
     }
   }