Fluffy state network tests and logging improvements (#2402)

* Improve tests by waiting for content to get into db.

* Improve state network logging.

fluffy/network/state/state_endpoints.nim

@@ -79,7 +79,7 @@ proc getAccountProof(
   while nibblesIdx < nibbles.len():
     let
       accountTrieNode = (await n.getAccountTrieNode(key)).valueOr:
-        # log something here
+        warn "Failed to get account trie node when building account proof"
         return Opt.none(TrieProof)
       trieNode = accountTrieNode.node
 
@@ -106,7 +106,7 @@ proc getStorageProof(
   while nibblesIdx < nibbles.len():
     let
       contractTrieNode = (await n.getContractTrieNode(key)).valueOr:
-        # log something here
+        warn "Failed to get contract trie node when building account proof"
         return Opt.none(TrieProof)
       trieNode = contractTrieNode.node
 
@@ -131,7 +131,7 @@ proc getAccount(
       warn "Failed to get account proof"
       return Opt.none(Account)
     account = accountProof.toAccount().valueOr:
-      warn "Failed to get account from accountProof"
+      error "Failed to get account from accountProof"
       return Opt.none(Account)
 
   Opt.some(account)
@@ -165,7 +165,7 @@ proc getStorageAt*(
       warn "Failed to get storage proof"
       return Opt.none(UInt256)
     slotValue = storageProof.toSlot().valueOr:
-      warn "Failed to get slot from storageProof"
+      error "Failed to get slot from storageProof"
       return Opt.none(UInt256)
 
   Opt.some(slotValue)

fluffy/network/state/state_gossip.nim

@@ -140,7 +140,7 @@ proc recursiveGossipOffer*(
     key: AccountTrieNodeKey,
     offer: AccountTrieNodeOffer,
 ) {.async: (raises: [CancelledError]).} =
-  asyncSpawn gossipOffer(p, srcNodeId, keyBytes, offerBytes, key, offer)
+  await gossipOffer(p, srcNodeId, keyBytes, offerBytes, key, offer)
 
   # root node, recursive gossip is finished
   if key.path.unpackNibbles().len() == 0:
@@ -151,7 +151,7 @@ proc recursiveGossipOffer*(
     (parentKey, parentOffer) = offer.withKey(key).getParent()
     parentKeyBytes = parentKey.toContentKey().encode()
 
-  asyncSpawn recursiveGossipOffer(
+  await recursiveGossipOffer(
     p, srcNodeId, parentKeyBytes, parentOffer.encode(), parentKey, parentOffer
   )
 
@@ -165,7 +165,7 @@ proc recursiveGossipOffer*(
     key: ContractTrieNodeKey,
     offer: ContractTrieNodeOffer,
 ) {.async: (raises: [CancelledError]).} =
-  asyncSpawn gossipOffer(p, srcNodeId, keyBytes, offerBytes, key, offer)
+  await gossipOffer(p, srcNodeId, keyBytes, offerBytes, key, offer)
 
   # root node, recursive gossip is finished
   if key.path.unpackNibbles().len() == 0:
@@ -176,6 +176,6 @@ proc recursiveGossipOffer*(
     (parentKey, parentOffer) = offer.withKey(key).getParent()
     parentKeyBytes = parentKey.toContentKey().encode()
 
-  asyncSpawn recursiveGossipOffer(
+  await recursiveGossipOffer(
     p, srcNodeId, parentKeyBytes, parentOffer.encode(), parentKey, parentOffer
   )

fluffy/network/state/state_network.nim

@@ -86,25 +86,26 @@ proc getContent(
     let contentFromDB = n.contentDB.get(contentId)
     if contentFromDB.isSome():
       let contentValue = V.decode(contentFromDB.get()).valueOr:
-        error "Unable to decode account trie node content value from database"
+        error "Unable to decode state content value from database"
         return Opt.none(V)
 
-      info "Fetched account trie node from database"
+      info "Fetched state content value from database"
       return Opt.some(contentValue)
 
   let
     contentLookupResult = (
       await n.portalProtocol.contentLookup(contentKeyBytes, contentId)
     ).valueOr:
+      warn "Failed fetching state content from the network"
       return Opt.none(V)
     contentValueBytes = contentLookupResult.content
 
   let contentValue = V.decode(contentValueBytes).valueOr:
-    error "Unable to decode account trie node content value from content lookup"
+    warn "Unable to decode state content value from content lookup"
     return Opt.none(V)
 
   validateRetrieval(key, contentValue).isOkOr:
-    error "Validation of retrieved content failed"
+    warn "Validation of retrieved state content failed"
     return Opt.none(V)
 
   n.portalProtocol.storeContent(contentKeyBytes, contentId, contentValueBytes)
@@ -134,7 +135,7 @@ proc getStateRootByBlockHash*(
     n: StateNetwork, hash: BlockHash
 ): Future[Opt[KeccakHash]] {.async: (raises: [CancelledError]).} =
   if n.historyNetwork.isNone():
-    warn "History network is not available. Unable to get state root by block hash"
+    warn "History network is not available"
     return Opt.none(KeccakHash)
 
   let header = (await n.historyNetwork.get().getVerifiedBlockHeader(hash)).valueOr:
@@ -174,7 +175,7 @@ proc processOffer*(
   )
   info "Offered content validated successfully", contentKeyBytes
 
-  asyncSpawn gossipOffer(
+  await gossipOffer(
     n.portalProtocol, maybeSrcNodeId, contentKeyBytes, contentValueBytes, contentKey,
     contentValue,
   )

fluffy/network/state/state_validation.nim

@@ -33,6 +33,7 @@ proc isValidNextNode(
 
   nextNode.hashEquals(nextHash)
 
+# TODO: Refactor this function to improve maintainability
 proc validateTrieProof*(
     expectedRootHash: Opt[KeccakHash],
     path: Nibbles,

fluffy/tests/state_network_tests/state_test_helpers.nim

@@ -167,7 +167,7 @@ proc stop*(sn: StateNode) {.async.} =
   sn.stateNetwork.stop()
   await sn.discoveryProtocol.closeWait()
 
-proc containsId*(sn: StateNode, contentId: ContentId): bool =
+proc containsId*(sn: StateNode, contentId: ContentId): bool {.inline.} =
   return sn.stateNetwork.contentDB.get(contentId).isSome()
 
 proc mockBlockHashToStateRoot*(
@@ -187,3 +187,11 @@ proc mockBlockHashToStateRoot*(
   sn.portalProtocol().storeContent(
     contentKeyBytes, contentId, SSZ.encode(blockHeaderWithProof)
   )
+
+proc waitUntilContentAvailable*(sn: StateNode, contentId: ContentId) {.async.} =
+  var waitCount = 0
+  while not sn.containsId(contentId):
+    await sleepAsync(10.milliseconds)
+    inc waitCount
+    if waitCount > 1000:
+      break

fluffy/tests/state_network_tests/test_state_endpoints_vectors.nim

@@ -71,7 +71,9 @@ procSuite "State Endpoints":
       )
 
       # wait for recursive gossip to complete
-      await sleepAsync(2000.milliseconds)
+      for node in testData.recursive_gossip:
+        let keyBytes = node.content_key.hexToSeqByte().ByteList
+        await stateNode2.waitUntilContentAvailable(toContentId(keyBytes))
 
       let
         address =
@@ -201,7 +203,9 @@ procSuite "State Endpoints":
       )
 
       # wait for recursive gossip to complete
-      await sleepAsync(1000.milliseconds)
+      for node in testData.recursive_gossip:
+        let keyBytes = node.content_key.hexToSeqByte().ByteList
+        await stateNode2.waitUntilContentAvailable(toContentId(keyBytes))
 
       let
         address = EthAddress.fromHex("0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2")
@@ -246,8 +250,8 @@ procSuite "State Endpoints":
         contentValue,
       )
 
-      # wait for recursive gossip to complete
+      # wait for gossip to complete
-      await sleepAsync(1000.milliseconds)
+      await stateNode2.waitUntilContentAvailable(contentId)
 
       let
         address = EthAddress.fromHex("0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2")

fluffy/tests/state_network_tests/test_state_gossip_gossipoffer_vectors.nim

@@ -76,9 +76,7 @@ procSuite "State Gossip - Gossip Offer":
       )
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       # check that the offer was received by the second state instance
       let res1 =
@@ -151,9 +149,7 @@ procSuite "State Gossip - Gossip Offer":
       )
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       # check that the offer was received by the second state instance
       let res1 = await stateNode2.stateNetwork.getContractTrieNode(
@@ -216,9 +212,7 @@ procSuite "State Gossip - Gossip Offer":
       )
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       # check that the offer was received by the second state instance
       let res1 =
@@ -280,7 +274,9 @@ procSuite "State Gossip - Gossip Offer":
       )
 
       # wait for recursive gossip to complete
-      await sleepAsync(1000.milliseconds)
+      for node in testData.recursive_gossip:
+        let keyBytes = node.content_key.hexToSeqByte().ByteList
+        await stateNode2.waitUntilContentAvailable(toContentId(keyBytes))
 
       # check that all nodes were received by both state instances
       for kv in testData.recursive_gossip:
@@ -357,7 +353,9 @@ procSuite "State Gossip - Gossip Offer":
       )
 
       # wait for recursive gossip to complete
-      await sleepAsync(1000.milliseconds)
+      for node in testData.recursive_gossip:
+        let keyBytes = node.content_key.hexToSeqByte().ByteList
+        await stateNode2.waitUntilContentAvailable(toContentId(keyBytes))
 
       # check that all nodes were received by both state instances
       for kv in testData.recursive_gossip:

fluffy/tests/state_network_tests/test_state_network_offercontent_vectors.nim

@@ -271,9 +271,7 @@ procSuite "State Network - Offer Content":
       check offerResult.isOk()
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       let getRes =
         await stateNode2.stateNetwork.getAccountTrieNode(contentKey.accountTrieNodeKey)
@@ -322,9 +320,7 @@ procSuite "State Network - Offer Content":
       check offerResult.isOk()
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       let getRes = await stateNode2.stateNetwork.getContractTrieNode(
         contentKey.contractTrieNodeKey
@@ -374,9 +370,7 @@ procSuite "State Network - Offer Content":
       check offerResult.isOk()
 
       # wait for offer to be processed by state node 2
-      while not stateNode2.stateNetwork.contentQueue.empty():
+      await stateNode2.waitUntilContentAvailable(contentId)
-        await sleepAsync(1.milliseconds)
-      await sleepAsync(100.milliseconds)
 
       let getRes =
         await stateNode2.stateNetwork.getContractCode(contentKey.contractCodeKey)