nimbus-eth1/fluffy/network/state/state_network.nim

# Fluffy
# Copyright (c) 2021-2024 Status Research & Development GmbH
# Licensed and distributed under either of
#   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
#   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.

import
  stew/results, chronos, chronicles,
  eth/common/eth_hash,
  eth/common,
  eth/p2p/discoveryv5/[protocol, enr],
  ../../database/content_db,
  ../wire/[portal_protocol, portal_stream, portal_protocol_config],
  ./state_content

logScope:
  topics = "portal_state"

const
  stateProtocolId* = [byte 0x50, 0x0A]

type StateNetwork* = ref object
  portalProtocol*: PortalProtocol
  contentDB*: ContentDB
  contentQueue*: AsyncQueue[(Opt[NodeId], ContentKeysList, seq[seq[byte]])]
  processContentLoop: Future[void]

func toContentIdHandler(contentKey: ByteList): results.Opt[ContentId] =
  ok(toContentId(contentKey))

func decodeKV*(contentKey: ByteList, contentValue: seq[byte]): Opt[(ContentKey, OfferContentValue)] =
  const empty = Opt.none((ContentKey, OfferContentValue))
  let
    key = contentKey.decode().valueOr:
      return empty
    value = case key.contentType:
      of unused:
        return empty
      of accountTrieNode:
        let val = decodeSsz(contentValue, AccountTrieNodeOffer).valueOr:
          return empty
        OfferContentValue(contentType: accountTrieNode, accountTrieNode: val)
      of contractTrieNode:
        let val = decodeSsz(contentValue, ContractTrieNodeOffer).valueOr:
          return empty
        OfferContentValue(contentType: contractTrieNode, contractTrieNode: val)
      of contractCode:
        let val = decodeSsz(contentValue, ContractCodeOffer).valueOr:
          return empty
        OfferContentValue(contentType: contractCode, contractCode: val)

  Opt.some((key, value))

func decodeValue*(contentKey: ContentKey, contentValue: seq[byte]): Opt[RetrievalContentValue] =
  const empty = Opt.none(RetrievalContentValue)
  let
    value = case contentKey.contentType:
      of unused:
        return empty
      of accountTrieNode:
        let val = decodeSsz(contentValue, AccountTrieNodeRetrieval).valueOr:
          return empty
        RetrievalContentValue(contentType: accountTrieNode, accountTrieNode: val)
      of contractTrieNode:
        let val = decodeSsz(contentValue, ContractTrieNodeRetrieval).valueOr:
          return empty
        RetrievalContentValue(contentType: contractTrieNode, contractTrieNode: val)
      of contractCode:
        let val = decodeSsz(contentValue, ContractCodeRetrieval).valueOr:
          return empty
        RetrievalContentValue(contentType: contractCode, contractCode: val)

  Opt.some(value)

proc validateAccountTrieNode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: RetrievalContentValue): bool =
    true

proc validateContractTrieNode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: RetrievalContentValue): bool =
    true

proc validateContractCode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: RetrievalContentValue): bool =
    true

proc validateContent*(
  n: StateNetwork,
  contentKey: ContentKey,
  contentValue: RetrievalContentValue): bool =
    case contentKey.contentType:
      of unused:
        warn "Received content with unused content type"
        false
      of accountTrieNode:
        validateAccountTrieNode(n, contentKey, contentValue)
      of contractTrieNode:
        validateContractTrieNode(n, contentKey, contentValue)
      of contractCode:
        validateContractCode(n, contentKey, contentValue)

proc getContent*(n: StateNetwork, key: ContentKey):
    Future[Opt[seq[byte]]] {.async.} =
  let
    keyEncoded = encode(key)
    contentId = toContentId(key)
    contentInRange = n.portalProtocol.inRange(contentId)

  # When the content id is in the radius range, try to look it up in the db.
  if contentInRange:
    let contentFromDB = n.contentDB.get(contentId)
    if contentFromDB.isSome():
      return contentFromDB

  let content = await n.portalProtocol.contentLookup(keyEncoded, contentId)

  if content.isNone():
    return Opt.none(seq[byte])

  let
    contentResult = content.get()
    decodedValue = decodeValue(key, contentResult.content).valueOr:
      error "Unable to decode offered Key/Value"
      return Opt.none(seq[byte])

  if not validateContent(n, key, decodedValue):
    return Opt.none(seq[byte])

  # When content is found on the network and is in the radius range, store it.
  if content.isSome() and contentInRange:
    # TODO Add poke when working on state network
    # TODO When working on state network, make it possible to pass different
    # distance functions to store content
    n.portalProtocol.storeContent(keyEncoded, contentId, contentResult.content)

  # TODO: for now returning bytes, ultimately it would be nice to return proper
  # domain types.
  return Opt.some(contentResult.content)

proc validateAccountTrieNode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: OfferContentValue): bool =
    true

proc validateContractTrieNode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: OfferContentValue): bool =
    true

proc validateContractCode(
  n: StateNetwork,
  key: ContentKey,
  contentValue: OfferContentValue): bool =
    true

proc validateContent*(
  n: StateNetwork,
  contentKey: ContentKey,
  contentValue: OfferContentValue): bool =
    case contentKey.contentType:
      of unused:
        warn "Received content with unused content type"
        false
      of accountTrieNode:
        validateAccountTrieNode(n, contentKey, contentValue)
      of contractTrieNode:
        validateContractTrieNode(n, contentKey, contentValue)
      of contractCode:
        validateContractCode(n, contentKey, contentValue)

proc recursiveGossipAccountTrieNode(
    p: PortalProtocol,
    maybeSrcNodeId: Opt[NodeId],
    decodedKey: ContentKey,
    decodedValue: AccountTrieNodeOffer
    ): Future[void] {.async.} =
      var
        nibbles = decodedKey.accountTrieNodeKey.path.unpackNibbles()
        proof = decodedValue.proof

      # When nibbles is empty this means the root node was received. Recursive
      # gossiping is finished.
      if nibbles.len() == 0:
        return

      discard nibbles.pop()
      discard (distinctBase proof).pop()
      let
        updatedValue = AccountTrieNodeOffer(
          proof: proof,
          blockHash: decodedValue.blockHash,
        )
        updatedNodeHash = keccakHash(distinctBase proof[^1])
        encodedValue = SSZ.encode(updatedValue)
        updatedKey = AccountTrieNodeKey(path: nibbles.packNibbles(), nodeHash: updatedNodeHash)
        encodedKey = ContentKey(accountTrieNodeKey: updatedKey, contentType: accountTrieNode).encode()

      await neighborhoodGossipDiscardPeers(
        p, maybeSrcNodeId, ContentKeysList.init(@[encodedKey]), @[encodedValue]
      )

proc recursiveGossipContractTrieNode(
    p: PortalProtocol,
    maybeSrcNodeId: Opt[NodeId],
    decodedKey: ContentKey,
    decodedValue: ContractTrieNodeOffer
    ): Future[void] {.async.} =
      return

proc gossipContent*(
    p: PortalProtocol,
    maybeSrcNodeId: Opt[NodeId],
    contentKey: ByteList,
    decodedKey: ContentKey,
    contentValue: seq[byte],
    decodedValue: OfferContentValue
    ): Future[void] {.async.} =
  case decodedKey.contentType:
    of unused:
      raiseAssert "Gossiping content with unused content type"
    of accountTrieNode:
      await recursiveGossipAccountTrieNode(p, maybeSrcNodeId, decodedKey, decodedValue.accountTrieNode)
    of contractTrieNode:
      await recursiveGossipContractTrieNode(p, maybeSrcNodeId, decodedKey, decodedValue.contractTrieNode)
    of contractCode:
      await p.neighborhoodGossipDiscardPeers(
        maybeSrcNodeId, ContentKeysList.init(@[contentKey]), @[contentValue]
      )

proc new*(
    T: type StateNetwork,
    baseProtocol: protocol.Protocol,
    contentDB: ContentDB,
    streamManager: StreamManager,
    bootstrapRecords: openArray[Record] = [],
    portalConfig: PortalProtocolConfig = defaultPortalProtocolConfig): T =

  let cq = newAsyncQueue[(Opt[NodeId], ContentKeysList, seq[seq[byte]])](50)

  let s = streamManager.registerNewStream(cq)

  let portalProtocol = PortalProtocol.new(
    baseProtocol, stateProtocolId,
    toContentIdHandler, createGetHandler(contentDB), s,
    bootstrapRecords, config = portalConfig)

  portalProtocol.dbPut = createStoreHandler(contentDB, portalConfig.radiusConfig, portalProtocol)

  return StateNetwork(
    portalProtocol: portalProtocol,
    contentDB: contentDB,
    contentQueue: cq
  )

proc processContentLoop(n: StateNetwork) {.async.} =
  try:
    while true:
      let (maybeSrcNodeId, contentKeys, contentValues) = await n.contentQueue.popFirst()
      for i, contentValue in contentValues:
        let
          contentKey = contentKeys[i]
          (decodedKey, decodedValue) = decodeKV(contentKey, contentValue).valueOr:
            error "Unable to decode offered Key/Value"
            continue
        if validateContent(n, decodedKey, decodedValue):
          let
            valueForRetrieval = decodedValue.offerContentToRetrievalContent().encode()
            contentId = n.portalProtocol.toContentId(contentKey).valueOr:
              error "Received offered content with invalid content key", contentKey
              continue

          n.portalProtocol.storeContent(contentKey, contentId, valueForRetrieval)
          info "Received offered content validated successfully", contentKey

          await gossipContent(
            n.portalProtocol,
            maybeSrcNodeId,
            contentKey,
            decodedKey,
            contentValue,
            decodedValue
          )
        else:
          error "Received offered content failed validation", contentKey
  except CancelledError:
    trace "processContentLoop canceled"

proc start*(n: StateNetwork) =
  info "Starting Portal execution state network",
    protocolId = n.portalProtocol.protocolId
  n.portalProtocol.start()

  n.processContentLoop = processContentLoop(n)

proc stop*(n: StateNetwork) =
  n.portalProtocol.stop()

  if not n.processContentLoop.isNil:
    n.processContentLoop.cancelSoon()