fix: sync large blocks without OOMing

[mcrakhman]

Overview

This PR introduces dynamic parameter calculation for the block sync reactor based on observed block sizes. The system adapts retry timeouts, pending request limits, and memory usage in real-time as blocks are synced, optimizing for both small and large block sizes.

Changes to unmarshalling and copying (for all reactors)

Unmarshalling of message protos now happens in the same goroutine, only then we send the result asynchronously to respective channel, before we copied the bytes in this goroutine and sent raw bytes to a reactor. This prevents us from allocating an intermediate buffer for each message and doing twice the work. This reduces the memory usage during blocksync.

Core Components

1. BlockStats (block_stats.go)

A circular buffer that maintains last 70 received block sizes.
When a new block size is added:

• If buffer is not full: adds the value and updates max
• If buffer is full: replaces the oldest value, updates sum, and recalculates max if necessary

2. Block Pool Parameters (pool.go)

Parameters that determine the buffer sizes and timeouts.

• reqLimit: Maximum concurrent requests per peer
• retryTimeout: Duration before retrying a failed block request

All calculations use the maximum block size from the block stats (worst-case planning) rather than the average. This ensures conservative resource allocation.

a. Concurrent requests to single peer

Determines how many concurrent block requests can be sent to a single peer. In current p2p scheme the responses from peer are received sequentially, so there is no multiplexing at the reactor level (though we can get messages from different reactors in parallel). That means that even if we request a lot of blocks in parallel, we will still get them sequentially. So logically the number of concurrent requests to peer should be higher for small blocks because peers send them fast, but lower for large blocks.

b. Retry timeouts

Determines how long to wait before retrying a timed-out block request. Again, should be smaller for small blocks, and larger for large ones.

// setting some value between maxReqLimit and minReqLimit to limit the number of requests we send to one 
// this is done to maintain peer diversity
pool.reqLimit = interpolate(blockSize, minBlockSizeBytes, maxBlockSizeBytes, maxReqLimit, minReqLimit)
// in the same way finding an adequate timeout given the block size
pool.retryTimeout = time.Second * time.Duration(interpolate(blockSize, minBlockSizeBytes, maxBlockSizeBytes, minRetrySeconds, maxRetrySeconds))

3. Parameter Recalculation

Parameters are recalculated in three scenarios:

After each block is received:

• Block size is added to the rotating buffer
• All parameters recalculate based on the new max block size

4. Block Request Flow

Each block height has a dedicated bpRequester that manages the request lifecycle:

1. Initial request: Sent to the peer returned by pickIncrAvailablePeer()

   • Picks peers with lowest current receive rate (most available)
   • Increments peer's pending counter

2. Timeout handling: If no block arrives within retryTimeout:

   • Requester calls redo() on itself
   • Resets internal state but keeps the same height
   • Sends a new request to a potentially different peer
   • Can retry multiple times until the block arrives

3. Block arrival: When AddBlock is called:

   • Validates the block came from the expected peer
   • Stores block and extended commit in the requester
   • Decrements peer's pending counter
   • Adds block size to the rotating buffer (triggers recalculation)

FLUPs

1. Reduce large peer buffers after the end of the block sync to free memory

2. Save blocks in another goroutine, to speed up the block sync

3. Add option to skip process proposal, this would give us roughly the same guarantees as state sync and speed up the block sync even more

Some more context is referenced here in the issue: #2594

[rach-id]

Can you please explain how this works? it would make reviewing easier, I'm a bit lost in this PR :D

[evan-forbes]

nice improvement!

do we have an issue for

Add option to skip process proposal, this would give us roughly the same guarantees as state sync and speed up the block sync even more

yet? if not can we write one with the reasoning for doing this with state sync

generally still thinking if there's something we can do that gives us the result we want without having to remember all these parameters and how they interact block by block in the scenario where we are debugging or tuning

only not approving to make sure my understanding is correct and to sleep on solutions

[rach-id]

Thanks for applying the feedback and adding the summary 🙏 🙏

Left a few nits and a few comments to understand more. Given this is related to blocksync, I'm being very careful with it.

[mcrakhman]

nice improvement!

do we have an issue for

Add option to skip process proposal, this would give us roughly the same guarantees as state sync and speed up the block sync even more

yet? if not can we write one with the reasoning for doing this with state sync

generally still thinking if there's something we can do that gives us the result we want without having to remember all these parameters and how they interact block by block in the scenario where we are debugging or tuning

only not approving to make sure my understanding is correct and to sleep on solutions

Added a reference to respective issue with regards to FLUPs. As for simplifying the constants, I tried doing that, but I didn't come up with a cleaner approach. You can see my calculations in the description to the PR. But basically because we calculate some value, we need to know min and max and then interpolate between those border values (we do this for both timeout and pending requests). We want to also have default values, when we didn't load enough blocks (edge case) etc. So maybe it is possible to get rid of a couple constants, but we will still have a lot due to the number of variables in our calculations.

[tzdybal]

This PR is huge and complex. The logic is very sophisticated.

I can see some changes that would deserve separate PRs, mostly for clarity:

1. Earlier deserialization that avoids copying large chunks of data.
2. Adding tracing.

I spend a lot of time reviewing this. The changes are not obvious, and it's a bit hard to reason about them. This raises my concern, that debugging any issues related to this code could also be hard. Claude code suggested that there might be some data races in the new code, but I'm not sure if they are just false-positives from AI review (but it's concerning anyways).
There is also a "memory leak" (accumulation of objects in map) - details in comment.

I don't fully understand (and there are no comments about) some code removals, like new height notification / second peer handling.

Proposed solution tries to be very comprehensive and data driven, but in my opinion some assumptions are not ideal. For example, next blocks can always be of max size, and we need to be able to handle this. Similarly, dynamic timeouts is a great idea, but without knowing the exact size of a block we need to prepare for the worst (which is IMHO max block size, not max size observed recently).

I'm not a fan of dropping requesters logic. If we're going to modify the already convoluted logic in blocksync/pool.go, I would prefer something like waiting before sendRequest (sync.Cond is first that comes to my mind) if there is no memory to receive next blocks.

[tzdybal]

Actual check for capacity would also be useful.

[tzdybal]

Why we're using max observed block size? IMHO this is not a worst case. Using max allowed block size is actually the worst case.

[mcrakhman]

Using max observed block will be more conservative and allow us to have larger timeouts for the future blocks, otherwise if we get some blocks at 0 and some blocks 128 we will have timeouts that would be to low for large blocks. We can also use median for that. But still, I don't want to complicate this right now

[tzdybal]

There is a InEpsilon function in testify for this.

[mcrakhman]

Removed this also

[tzdybal]

This is the only place we delete from droppedRequesters. If we won't get those "unwanted" blocks from them, we leak / accumulate droppedRequesters forever (I know it's just struct{} but it's still a leak).

[tzdybal]

Some comments lost while submitting 😭 For some reason even typing lags for me on this PR...
Fortunately the main description is there at least...

[rach-id]

Final round of review.

Thanks for applying the previous feedback. I agree with the direction of the PR and with the proposed solution, given how hard it is to estimate how many requesters to use without knowing the block sizes that will be requested.

The feedback I left is implementation related to make it easy for future us to debug in case of issues.

Also, thanks for your patience with the multiple feedback rounds 🙏 🙏

[evan-forbes]

nice simplification!!

I appreciate all the refactors here

[evan-forbes]

I think we can dismiss the block @tzdybal is that correct?

Out of date