[ETCM-211] Tracking of headers/bodies requests that will be ignored due to invalidation #749
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,6 +9,10 @@ import cats.instances.future._ | |
| import cats.instances.option._ | ||
| import cats.syntax.either._ | ||
| import io.iohk.ethereum.blockchain.sync.PeersClient._ | ||
| import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{ | ||
| AwaitingBodiesToBeIgnored, | ||
| AwaitingHeadersToBeIgnored | ||
| } | ||
| import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.{ImportNewBlock, NotOnTop, OnTop} | ||
| import io.iohk.ethereum.crypto.kec256 | ||
| import io.iohk.ethereum.domain._ | ||
|
|
@@ -91,27 +95,51 @@ class BlockFetcher( | |
|
|
||
| private def handleHeadersMessages(state: BlockFetcherState): Receive = { | ||
| case Response(peer, BlockHeaders(headers)) if state.isFetchingHeaders => | ||
| val newState = | ||
| if (state.fetchingHeadersState == AwaitingHeadersToBeIgnored) { | ||
|
There was a problem hiding this comment. Maybe I am missing somehing, but what will happen in case of request time out ?
Is it correct, or there is some detail i am missing ? There was a problem hiding this comment. You are right! I thought I had analyzed that but I missed it I'll update this PR with something even closer to what we do when receiving the response |
||
| log.debug( | ||
| "Received {} headers starting from block {} that will be ignored", | ||
| headers.size, | ||
| headers.headOption.map(_.number) | ||
| ) | ||
| state.withHeaderFetchReceived | ||
| } else { | ||
| log.debug("Fetched {} headers starting from block {}", headers.size, headers.headOption.map(_.number)) | ||
|
|
||
| state.validatedHeaders(headers) match { | ||
| case Left(err) => | ||
| peersClient ! BlacklistPeer(peer.id, err) | ||
| state.withHeaderFetchReceived | ||
| case Right(validHeaders) => | ||
| state.withHeaderFetchReceived.appendHeaders(validHeaders) | ||
| } | ||
| } | ||
|
|
||
| fetchBlocks(newState) | ||
| case RetryHeadersRequest if state.isFetchingHeaders => | ||
| log.debug("Time-out occurred while waiting for headers") | ||
|
There was a problem hiding this comment. more of discussion type comment. Do you think it would bring problems if we would receive this time out response later ? i.e something like:
Do you think it could result in something more than blacklising peer ? (for example triggering this bug with unbounded resources usage) There was a problem hiding this comment. Definitely, we are currently doing no validations that the headers received match the request With that sort of validations we should be safe I think, but for now I was assuming that that case will never happen, the 30 seconds till timeout should be more than enough to prevent that, right? There was a problem hiding this comment. That makes me think if this whole solution shouldn't be implemented as Actor/class in between fetcher and peers client. It could track requests made and invalidate previous ones basing on strategy choosen (at-most-n, allow-all, etc.). There was a problem hiding this comment. What would this actor do? Are you maybe thinking of something like the BlockChainDataFetcher of our other project? (only on how it requests headers and bodies)
That at least would focus on ☝️ part and allow validating that the responses received match the requests we sent Either way I'd do that effort in parallel with this kind of patching that only touches the minimum required (and is needed for the testnet) I'm a bit worried of how risky any design changes on our sync could be... though that BlockChainDataFetcher refactoring would be low risk I think There was a problem hiding this comment. Well, Where There was a problem hiding this comment. so imo details up to discussion, but i am all for doing some kinda detailed request-response tracking. Unfortunately as eth do not have any request-id for its request-response messages, such thing is ultimately necessary if we want to have possiblitly to cancel in fly requests. There was a problem hiding this comment.
I just realized this might not be a problem, our PeerRequestHandler awaiting the response should have been killed by then so any too long into the future response should be discarded. Either way I added this task to add validations that the request/response match and then detect malicious peer behaviour: https://jira.iohk.io/browse/ETCM-283
Due to our PeerRequestHandler design this might be easier to do 🤔 we can identify the request by the actor that's in charge of getting the response. If we use a single peer per request type that might work (for now we don't have even more than one request type in parallel) |
||
|
|
||
| val newState = state.withHeaderFetchReceived | ||
| fetchBlocks(newState) | ||
| } | ||
|
|
||
| private def handleBodiesMessages(state: BlockFetcherState): Receive = { | ||
| case Response(peer, BlockBodies(bodies)) if state.isFetchingBodies => | ||
| val newState = | ||
| if (state.fetchingBodiesState == AwaitingBodiesToBeIgnored) { | ||
| log.debug("Received {} block bodies that will be ignored", bodies.size) | ||
| state.withBodiesFetchReceived | ||
| } else { | ||
| log.debug("Fetched {} block bodies", bodies.size) | ||
| state.withBodiesFetchReceived.addBodies(peer, bodies) | ||
| } | ||
|
|
||
| fetchBlocks(newState) | ||
| case RetryBodiesRequest if state.isFetchingBodies => | ||
| log.debug("Time-out occurred while waiting for bodies") | ||
|
|
||
| val newState = state.withBodiesFetchReceived | ||
| fetchBlocks(newState) | ||
| } | ||
|
|
||
| private def handleStateNodeMessages(state: BlockFetcherState): Receive = { | ||
|
|
@@ -207,7 +235,7 @@ class BlockFetcher( | |
| .filter(!_.hasFetchedTopHeader) | ||
| .filter(!_.hasReachedSize(syncConfig.maxFetcherQueueSize)) | ||
| .tap(fetchHeaders) | ||
| .map(_.withNewHeadersFetch) | ||
| .getOrElse(fetcherState) | ||
|
|
||
| private def fetchHeaders(state: BlockFetcherState): Unit = { | ||
|
|
@@ -229,10 +257,12 @@ class BlockFetcher( | |
| .filter(!_.isFetchingBodies) | ||
| .filter(_.waitingHeaders.nonEmpty) | ||
| .tap(fetchBodies) | ||
| .map(state => state.withNewBodiesFetch) | ||
| .getOrElse(fetcherState) | ||
|
|
||
| private def fetchBodies(state: BlockFetcherState): Unit = { | ||
| log.debug("Fetching bodies") | ||
|
|
||
| val hashes = state.takeHashes(syncConfig.blockBodiesPerRequest) | ||
| requestBlockBodies(hashes) pipeTo self | ||
| } | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package io.iohk.ethereum | ||
|
|
||
| import akka.util.ByteString | ||
| import io.iohk.ethereum.domain.Block | ||
|
|
||
| object BlockHelpers { | ||
|
|
||
| def generateChain(amount: Int, parent: Block): Seq[Block] = { | ||
| (1 to amount).foldLeft[Seq[Block]](Nil){ case (acc, _) => | ||
| val baseBlock = Fixtures.Blocks.ValidBlock.block | ||
|
|
||
| val parentHeader = acc.lastOption.getOrElse(parent) | ||
| val blockHeader = baseBlock.header.copy( | ||
| number = parentHeader.number + 1, | ||
| parentHash = parentHeader.hash, | ||
| // Random nonce used for having our blocks be different | ||
| nonce = ByteString(Math.random().toString) | ||
| ) | ||
| val block = baseBlock.copy(header = blockHeader) | ||
|
|
||
| acc :+ block | ||
| } | ||
| } | ||
|
|
||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'd move this code into method on
BlockFetcherStateand leave here only loggingThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As a handleHeaderResponse function you mean? Isn't the current structure more oriented to doing the handling of responses on the actor? Shouldn't we move the bodies/nodes response handling as well?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
After second look - let's keep it as is. It's just - the more I work with actors, more I want extract all the logic from them so it can be tested without creating that actor.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
it is totally my current sentiments about actors, they should have as little logic as possible and should be only thin communication layer over main logic.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, I'm having the same concerns with them
However for this case in particular I think we would fall into a too big and not understandable BlockFetcherState class if we start moving everything there, we should probably start splitting up this logic into other classes