proxy: log filters (#3461)

* txs, receipts and logs

* format

* add log filters

* lint and rebase fixes

* resolve filters

* remove bugs

* format

* fixes

* trigger CI

* review

* revert line

Author: chirag-parmar

execution_chain/version.nim

@@ -18,7 +18,10 @@ const
 
 static:
   doAssert(nimbusRevision.len == 8, "nimbusRevision must consist of 8 characters")
-  doAssert(nimbusRevision.allIt(it in HexDigits), "nimbusRevision should contains only hex chars")
+  doAssert(
+    nimbusRevision.allIt(it in HexDigits),
+    "nimbusRevision should contains only hex chars",
+  )
 
 proc gitFolderExists(path: string): bool {.compileTime.} =
   # walk up parent folder to find `.git` folder

nimbus_verified_proxy/rpc/blocks.nim

@@ -23,26 +23,26 @@ import
 
 proc resolveBlockTag*(
     vp: VerifiedRpcProxy, blockTag: BlockTag
-): Result[base.BlockNumber, string] =
+): Result[BlockTag, string] =
   if blockTag.kind == bidAlias:
     let tag = blockTag.alias.toLowerAscii()
     case tag
     of "latest":
       let hLatest = vp.headerStore.latest.valueOr:
         return err("Couldn't get the latest block number from header store")
-      ok(hLatest.number)
+      ok(BlockTag(kind: bidNumber, number: Quantity(hLatest.number)))
     of "finalized":
       let hFinalized = vp.headerStore.finalized.valueOr:
         return err("Couldn't get the latest block number from header store")
-      ok(hFinalized.number)
+      ok(BlockTag(kind: bidNumber, number: Quantity(hFinalized.number)))
     of "earliest":
       let hEarliest = vp.headerStore.earliest.valueOr:
         return err("Couldn't get the latest block number from header store")
-      ok(hEarliest.number)
+      ok(BlockTag(kind: bidNumber, number: Quantity(hEarliest.number)))
     else:
       err("No support for block tag " & $blockTag)
   else:
-    ok(base.BlockNumber(distinctBase(blockTag.number)))
+    ok(blockTag)
 
 func convHeader*(blk: eth_api_types.BlockObject): Header =
   let nonce = blk.nonce.valueOr:
@@ -184,10 +184,8 @@ proc getBlock*(
 proc getBlock*(
     vp: VerifiedRpcProxy, blockTag: BlockTag, fullTransactions: bool
 ): Future[Result[BlockObject, string]] {.async.} =
-  let
-    n = vp.resolveBlockTag(blockTag).valueOr:
-      return err(error)
-    numberTag = BlockTag(kind: BlockIdentifierKind.bidNumber, number: Quantity(n))
+  let numberTag = vp.resolveBlockTag(blockTag).valueOr:
+    return err(error)
 
   # get the target block
   let blk =
@@ -196,7 +194,7 @@ proc getBlock*(
     except CatchableError as e:
       return err(e.msg)
 
-  if n != distinctBase(blk.number):
+  if numberTag.number != blk.number:
     return
       err("the downloaded block number doesn't match with the requested block number")
 
@@ -235,9 +233,9 @@ proc getHeader*(
     vp: VerifiedRpcProxy, blockTag: BlockTag
 ): Future[Result[Header, string]] {.async.} =
   let
-    n = vp.resolveBlockTag(blockTag).valueOr:
+    numberTag = vp.resolveBlockTag(blockTag).valueOr:
       return err(error)
-    numberTag = BlockTag(kind: BlockIdentifierKind.bidNumber, number: Quantity(n))
+    n = distinctBase(numberTag.number)
     cachedHeader = vp.headerStore.get(n)
 
   if cachedHeader.isNone():

nimbus_verified_proxy/rpc/receipts.nim

@@ -29,7 +29,6 @@ func toReceipt(rec: ReceiptObject): Receipt =
   let isHash = not rec.status.isSome()
 
   let status = rec.status.isSome() and rec.status.get() == 1.Quantity
-
   return Receipt(
     hash: rec.transactionHash,
     isHash: isHash,
@@ -51,7 +50,6 @@ proc getReceipts(
       await vp.rpcClient.eth_getBlockReceipts(blockTag)
     except CatchableError as e:
       return err(e.msg)
-
   if rxs.isSome():
     if orderedTrieRoot(toReceipts(rxs.get())) != header.receiptsRoot:
       return
@@ -86,15 +84,32 @@ proc getReceipts*(
 
   await vp.getReceipts(header, numberTag)
 
-proc getLogs*(
-    vp: VerifiedRpcProxy, filterOptions: FilterOptions
-): Future[Result[seq[LogObject], string]] {.async.} =
-  let logObjs =
-    try:
-      await vp.rpcClient.eth_getLogs(filterOptions)
-    except CatchableError as e:
-      return err(e.msg)
+proc resolveFilterTags*(
+    vp: VerifiedRpcProxy, filter: FilterOptions
+): Result[FilterOptions, string] =
+  if filter.blockHash.isSome():
+    return ok(filter)
+  let
+    fromBlock = filter.fromBlock.get(types.BlockTag(kind: bidAlias, alias: "latest"))
+    toBlock = filter.toBlock.get(types.BlockTag(kind: bidAlias, alias: "latest"))
+    fromBlockNumberTag = vp.resolveBlockTag(fromBlock).valueOr:
+      return err(error)
+    toBlockNumberTag = vp.resolveBlockTag(toBlock).valueOr:
+      return err(error)
 
+  return ok(
+    FilterOptions(
+      fromBlock: Opt.some(fromBlockNumberTag),
+      toBlock: Opt.some(toBlockNumberTag),
+      address: filter.address,
+      topics: filter.topics,
+      blockHash: filter.blockHash,
+    )
+  )
+
+proc verifyLogs*(
+    vp: VerifiedRpcProxy, filter: FilterOptions, logObjs: seq[LogObject]
+): Future[Result[void, string]] {.async.} =
   # store block hashes contains the logs so that we can batch receipt requests
   var
     prevBlockHash: Hash32
@@ -109,7 +124,6 @@ proc getLogs*(
         rxs = (await vp.getReceipts(lg.blockHash.get())).valueOr:
           return err(error)
         prevBlockHash = lg.blockHash.get()
-
       let
         txIdx = distinctBase(lg.transactionIndex.get())
         logIdx =
@@ -119,7 +133,25 @@ proc getLogs*(
 
       if rxLog.address != lg.address or rxLog.data != lg.data or
           rxLog.topics != lg.topics or
-          (not match(toLog(lg), filterOptions.address, filterOptions.topics)):
+          lg.blockNumber.get() < filter.fromBlock.get().number or
+          lg.blockNumber.get() > filter.toBlock.get().number or
+          (not match(toLog(lg), filter.address, filter.topics)):
         return err("one of the returned logs is invalid")
 
+  ok()
+
+proc getLogs*(
+    vp: VerifiedRpcProxy, filter: FilterOptions
+): Future[Result[seq[LogObject], string]] {.async.} =
+  let
+    resolvedFilter = vp.resolveFilterTags(filter).valueOr:
+      return err(error)
+    logObjs =
+      try:
+        await vp.rpcClient.eth_getLogs(resolvedFilter)
+      except CatchableError as e:
+        return err(e.msg)
+
+  ?(await vp.verifyLogs(resolvedFilter, logObjs))
+
   return ok(logObjs)

nimbus_verified_proxy/rpc/rpc_eth_api.nim

@@ -11,6 +11,7 @@ import
   results,
   chronicles,
   stew/byteutils,
+  nimcrypto/sysrand,
   json_rpc/[rpcserver, rpcclient, rpcproxy],
   eth/common/accounts,
   web3/eth_api,
@@ -237,6 +238,7 @@ proc installEthApiHandlers*(vp: VerifiedRpcProxy) =
         await vp.rpcClient.eth_getTransactionByHash(txHash)
       except CatchableError as e:
         raise newException(ValueError, e.msg)
+
     if tx.hash != txHash:
       raise newException(
         ValueError,
@@ -274,7 +276,86 @@ proc installEthApiHandlers*(vp: VerifiedRpcProxy) =
     (await vp.getLogs(filterOptions)).valueOr:
       raise newException(ValueError, error)
 
-  # TODO:
+  vp.proxy.rpc("eth_newFilter") do(filterOptions: FilterOptions) -> string:
+    if vp.filterStore.len >= MAX_FILTERS:
+      raise newException(ValueError, "FilterStore already full")
+
+    var
+      id: array[8, byte] # 64bits
+      strId: string
+
+    for i in 0 .. (MAX_ID_TRIES + 1):
+      if randomBytes(id) != len(id):
+        raise newException(
+          ValueError, "Couldn't generate a random identifier for the filter"
+        )
+
+      strId = toHex(id)
+
+      if not vp.filterStore.contains(strId):
+        break
+
+      if i >= MAX_ID_TRIES:
+        raise
+          newException(ValueError, "Couldn't create a unique identifier for the filter")
+
+    vp.filterStore[strId] =
+      FilterStoreItem(filter: filterOptions, blockMarker: Opt.none(Quantity))
+
+    return strId
+
+  vp.proxy.rpc("eth_uninstallFilter") do(filterId: string) -> bool:
+    if filterId in vp.filterStore:
+      vp.filterStore.del(filterId)
+      return true
+
+    return false
+
+  vp.proxy.rpc("eth_getFilterLogs") do(filterId: string) -> seq[LogObject]:
+    if filterId notin vp.filterStore:
+      raise newException(ValueError, "Filter doesn't exist")
+
+    (await vp.getLogs(vp.filterStore[filterId].filter)).valueOr:
+      raise newException(ValueError, error)
+
+  vp.proxy.rpc("eth_getFilterChanges") do(filterId: string) -> seq[LogObject]:
+    if filterId notin vp.filterStore:
+      raise newException(ValueError, "Filter doesn't exist")
+
+    let
+      filterItem = vp.filterStore[filterId]
+      filter = vp.resolveFilterTags(filterItem.filter).valueOr:
+        raise newException(ValueError, error)
+      # after resolving toBlock is always some and a number tag
+      toBlock = filter.toBlock.get().number
+
+    if filterItem.blockMarker.isSome() and toBlock <= filterItem.blockMarker.get():
+      raise newException(ValueError, "No changes for the filter since the last query")
+
+    let
+      fromBlock =
+        if filterItem.blockMarker.isSome():
+          Opt.some(
+            types.BlockTag(kind: bidNumber, number: filterItem.blockMarker.get())
+          )
+        else:
+          filter.fromBlock
+
+      changesFilter = FilterOptions(
+        fromBlock: fromBlock,
+        toBlock: filter.toBlock,
+        address: filter.address,
+        topics: filter.topics,
+        blockHash: filter.blockHash,
+      )
+      logObjs = (await vp.getLogs(changesFilter)).valueOr:
+        raise newException(ValueError, error)
+
+    # all logs verified so we can update blockMarker
+    vp.filterStore[filterId].blockMarker = Opt.some(toBlock)
+
+    return logObjs
+
   # Following methods are forwarded directly to the web3 provider and therefore
   # are not validated in any way.
   vp.proxy.registerProxyMethod("net_version")

nimbus_verified_proxy/rpc_api_backend.nim

@@ -7,7 +7,12 @@
 
 {.push raises: [], gcsafe.}
 
-import json_rpc/[rpcproxy, rpcclient], web3/[eth_api, eth_api_types], stint, ./types
+import
+  json_rpc/[rpcproxy, rpcclient],
+  web3/[eth_api, eth_api_types],
+  stint,
+  std/json,
+  ./types
 
 proc initNetworkApiBackend*(vp: VerifiedRpcProxy): EthApiBackend =
   let

nimbus_verified_proxy/tests/test_receipts.nim

@@ -92,3 +92,73 @@ suite "test receipts verification":
       ts.loadLogs(filterOptions, logs)
       let verifiedLogs = waitFor vp.proxy.getClient().eth_getLogs(filterOptions)
       check verifiedLogs.len == logs.len
+
+  test "create filters and uninstall filters":
+    # filter options without any tags would test resolving default "latest"
+    let filterOptions = FilterOptions(
+      topics:
+        @[
+          TopicOrList(
+            kind: SingleOrListKind.slkSingle,
+            single:
+              bytes32"0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+          )
+        ],
+      blockHash: Opt.none(Hash32),
+    )
+
+    let
+      # create a filter
+      newFilter = waitFor vp.proxy.getClient().eth_newFilter(filterOptions)
+      # deleting will prove if the filter was created
+      delStatus = waitFor vp.proxy.getClient().eth_uninstallFilter(newFilter)
+
+    check delStatus
+
+    let
+      unknownFilterId = "thisisacorrectfilterid"
+      delStatus2 = waitFor vp.proxy.getClient().eth_uninstallFilter(newFilter)
+
+    check not delStatus2
+
+  test "get logs using filter changes":
+    let
+      blk = getBlockFromJson("nimbus_verified_proxy/tests/data/Paris.json")
+      rxs = getReceiptsFromJson("nimbus_verified_proxy/tests/data/receipts.json")
+      logs = getLogsFromJson("nimbus_verified_proxy/tests/data/logs.json")
+
+    # update block tags because getLogs (uses)-> getReceipts (uses)-> getHeader
+    ts.loadBlockReceipts(blk, rxs)
+    discard vp.headerStore.add(convHeader(blk), blk.hash)
+    discard vp.headerStore.updateFinalized(convHeader(blk), blk.hash)
+
+    # filter options without any tags would test resolving default "latest"
+    let filterOptions = FilterOptions(
+      topics:
+        @[
+          TopicOrList(
+            kind: SingleOrListKind.slkSingle,
+            single:
+              bytes32"0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+          )
+        ],
+      blockHash: Opt.none(Hash32),
+    )
+
+    ts.loadLogs(filterOptions, logs)
+
+    let
+      # create a filter
+      newFilter = waitFor vp.proxy.getClient().eth_newFilter(filterOptions)
+      filterLogs = waitFor vp.proxy.getClient().eth_getFilterLogs(newFilter)
+      filterChanges = waitFor vp.proxy.getClient().eth_getFilterChanges(newFilter)
+
+    check filterLogs.len == logs.len
+    check filterChanges.len == logs.len
+
+    try:
+      let againFilterChanges =
+        waitFor vp.proxy.getClient().eth_getFilterChanges(newFilter)
+      check false
+    except CatchableError as e:
+      check true

nimbus_verified_proxy/types.nim

@@ -8,6 +8,7 @@
 {.push raises: [], gcsafe.}
 
 import
+  std/tables,
   json_rpc/[rpcproxy, rpcclient],
   web3/[eth_api, eth_api_types],
   stint,
@@ -21,6 +22,8 @@ const
   ACCOUNTS_CACHE_SIZE = 128
   CODE_CACHE_SIZE = 64
   STORAGE_CACHE_SIZE = 256
+  MAX_ID_TRIES* = 10
+  MAX_FILTERS* = 256
 
 type
   AccountsCacheKey* = (Root, Address)
@@ -63,6 +66,10 @@ type
     eth_getTransactionByHash*: GetTransactionByHashProc
     eth_getLogs*: GetLogsProc
 
+  FilterStoreItem* = object
+    filter*: FilterOptions
+    blockMarker*: Opt[Quantity]
+
   VerifiedRpcProxy* = ref object
     evm*: AsyncEvm
     proxy*: RpcProxy
@@ -74,6 +81,7 @@ type
 
     # TODO: when the list grows big add a config object instead
     # config parameters
+    filterStore*: Table[string, FilterStoreItem]
     chainId*: UInt256
     maxBlockWalk*: uint64