Decouple blobs

[emhane]

Issue Addressed

sigp#3991

Proposed Changes

A block with kzg commits is not available until its blobs have arrived.
Save kzg commits against blobs verification until after execution payload verification, along with AvailabilityPendingBlock.

Additional Info

A block's execution payload has to be verified at some point before importing to fork choice. If a block sets of to get this done after it gets gossip verified and propagated, then in some cases we may be lucky that, while waiting on input from execution layer, all of our blobs came. If this is not the case, we cache the ExecutedBlock which holds a DataAvailabilityHandle, and put the sender to the DataAvailabilityHandle in the blobs cache.

Building on from this solution

lighthouse/beacon_node/beacon_chain/src/blob_verification.rs

Line 275 in dd31621

type DataAvailabilityHandle<E> = JoinHandle<Result<Option<Arc<BlobsSidecar<E>>>, BlobError>>;

I'm keeping in mind that the probability that the blocks always comes last, after all its blobs, is small, and I'm allowing therefore an efficient solution for most scenarios at the cost of a less efficient solution for the scenario that the block comes last, here:
https://github.com/emhane/lighthouse/blob/97a1847e599fccd9c8ac6cc7ba5c301e4f554b1f/beacon_node/network/src/beacon_processor/worker/gossip_methods.rs#L665-L692

[realbigsean]

This is a way to avoid the block-blobs group being a future to poll, and rather move it into chain level on input from gossip, with the benefits of skipping double execution payload verification in most cases. Let me know your thoughts.

I'm confused how a double execution payload verification is avoided? I think the use of pushed-based processing on every blob is what creates a risk of duplicate work whereas polling a future that completes on blob verification success/failure makes it pretty easily avoided.

It's probably not clear what I was thinking on this branch cause it's a bit incomplete so far so here's some more detail on the direction:

Generally the idea is that any block we get that passes gossip verification should be an AvailabilityPendingBlock and we should just process as much of it as we can as normal, including payload verification. The future in the availability pending block should complete under conditions that are something like this:

1. All blobs have been received and KZG verification against the block passes
2. A timeout is hit, whatever we deem reasonable to consider data "unavailable"

The first part of this future can be a oneshot::channel's Receiver. The Sender can be stored in the blobs cache. Every time we get a block or blob on gossip, we can check if we have all components necessary to verify data availability, and if so, verify it and send a message on the Sender and evict it from the cache.

This is an example of what the cache could look like:

BlobCacheKey {
  breacon_block_root: Hash256,
}

struct BlobCacheValue {
  block: Option<Arc<SignedBeaconBlock>>, 
  received_blobs: Vec<BlobSidecar>`,
  verify_kzg_tx: Sender,
}

Note that a cache here is probably what we want not a buffer, because we also need to draw from this when both serving RPC requests for blocks/blobs and in order to know which blobs we are lacking to trigger the right RPC requests for blocks/blobs.

We'll want to trigger a blob by root parent lookup and queueing for reprocessing somewhere but I don't know where the best place for that is at this point.

[emhane]

This is a way to avoid the block-blobs group being a future to poll, and rather move it into chain level on input from gossip, with the benefits of skipping double execution payload verification in most cases. Let me know your thoughts.

I'm confused how a double execution payload verification is avoided? I think the use of pushed-based processing on every blob is what creates a risk of duplicate work whereas polling a future that completes on blob verification success/failure makes it pretty easily avoided.

It's probably not clear what I was thinking on this branch cause it's a bit incomplete so far so here's some more detail on the direction:

Generally the idea is that any block we get that passes gossip verification should be an AvailabilityPendingBlock and we should just process as much of it as we can as normal, including payload verification. The future in the availability pending block should complete under conditions that are something like this:

1. All blobs have been received and KZG verification against the block passes
2. A timeout is hit, whatever we deem reasonable to consider data "unavailable"

The first part of this future can be a oneshot::channel's Receiver. The Sender can be stored in the blobs cache. Every time we get a block or blob on gossip, we can check if we have all components necessary to verify data availability, and if so, verify it and send a message on the Sender and evict it from the cache.

This is an example of what the cache could look like:

BlobCacheKey {
  breacon_block_root: Hash256,
}

struct BlobCacheValue {
  block: Option<Arc<SignedBeaconBlock>>, 
  received_blobs: Vec<BlobSidecar>`,
  verify_kzg_tx: Sender,
}

Note that a cache here is probably what we want not a buffer, because we also need to draw from this when both serving RPC requests for blocks/blobs and in order to know which blobs we are lacking to trigger the right RPC requests for blocks/blobs.

We'll want to trigger a blob by root parent lookup and queueing for reprocessing somewhere but I don't know where the best place for that is at this point.

Yeah sure we can do that. I was thinking smthg like this, when the block-blobs group has already been down the rabbit hole once, we advance_state on the buffer item.

pub struct BlockBufferItem<T: BeaconChainTypes> {
    block_root: Hash256,
    block: BufferBlock<T>,
}

pub enum BufferBlock<T: BeaconChainTypes> {
    GossipVerified(GossipVerifiedBlock<T>),
    PayloadExecuted(ExecutedBlock<T>),
}

impl<T: BeaconChainTypes> BlockBufferItem<T> {
    pub fn new(block_root: Hash256, gossip_verified_block: GossipVerifiedBlock<T>) -> Self {
        BlockBufferItem{
            block_root,
            block: BufferBlock::GossipVerified(gossip_verified_block)
        }
    }

    pub fn advance_state(&mut self, block: ExecutedBlock<T>) {
        if BufferBlock::GossipVerified(_) = self.block {
            *self.block = BufferBlock::PayloadExecuted(block)
        }
    }
}

I see 2 problems with your solution:

1. verifying the kzg-commits against transactions before the execution payload has been verified.
2. what is the halting criteria to know when to verify kzg-commits? as it should happen after payload verification and the possibility that we get more than one blob pointing to the same index or simply a wrong blob exists, and how will this block-blob group be in place to receive more blobs if the correct blobs arrive while it is in kzg-verification?

[realbigsean]

1. verifying the kzg-commits against transactions before the execution payload has been verified.

I don't think this necessarily needs to happen before payload verification, I think we can look at them as independent. For this, I think we'll need to add a bool to ConsensusContext to know whether or not we've made this check or should make this check in per_block_processing

2. what is the halting criteria to know when to verify kzg-commits? as it should happen after payload verification and the possibility that we get more than one blob pointing to the same index or simply a wrong blob exists, and how will this block-blob group be in place to receive more blobs if the correct blobs arrive while it is in kzg-verification?

I don't think the kzg commit verification needs to happen after payload verification I think the two can happen in parallel. We'll have to pay close attention to how these invalid blob scenarios are described in gossip because only blobs passing all gossip conditions should be put in this cache:

https://github.com/ethereum/consensus-specs/pull/3244/files#diff-21cd1dde2a4815e714ef42d0cefa1bbd2997a4358ecf6e9e51025332c1d642fdR115-R116

The described scenarios would require the proposer to sign over invalid data so it's very likely this block gets orphaned.. but this is the type of thing we'll need to think very carefully about with regard to fork choice. Within a slot attesters should not be voting on unavailable block/blobs but we should have the ability to re-org to blocks that were not immediately available but became available to us and the rest of the network had available. (like if we later complete a by roots request for blobs)

Generally the halting criteria is the criteria for the completion of the future in the availability pending block. And this will have to be constrained by a timeout because at some point in block processing we will need to consider the data "unavailable". But this doesn't necessarily mean we should get rid of the block + blob for the above reasons. A failure of some types may mean we have to queue for reprocessing (waiting on the last blob).

[emhane]

So, @realbigsean I reset that first commit and used the future in a block cache item to know when all blobs have arrived, as you mentioned. I didn't see it necessary for this DataAvailabilityHandle to be spawned on a new thread, and rather think we should use one background thread, like we do for migrator, to loop through DataAvailabilityHandle futures. Furthermore, the reason I don't store the blobs with the AvailabilityPendingBlock when they come, is because I don't see the need and I want to put a lock on blob cache items. Blob cache items will contain a list of all blobs received over network for a block root. We could have one container cache item per block root where we put the block and its blobs, which ever comes first, if you prefer.

[emhane]

1. verifying the kzg-commits against transactions before the execution payload has been verified.

I don't think this necessarily needs to happen before payload verification, I think we can look at them as independent. For this, I think we'll need to add a bool to ConsensusContext to know whether or not we've made this check or should make this check in per_block_processing

If we want to parallelise the verification work, we can add an Arc to a block to a blob cache item at the start of process_block, so that when the expected number of blobs have arrived, we can spawn a blocking task and verify the kzg commits against the blocks transactions (optimistic that the execution payload is correct, which as you mention, there is stronger incentive for it to be than not to be). What is more economic here, verify the blobs one by one as they come over the network (after the block has come) with verify_kzg_proof or wait till they have all arrived (and the block) and use verify_aggregate_kzg_proof?

2. what is the halting criteria to know when to verify kzg-commits? as it should happen after payload verification and the possibility that we get more than one blob pointing to the same index or simply a wrong blob exists, and how will this block-blob group be in place to receive more blobs if the correct blobs arrive while it is in kzg-verification?

I don't think the kzg commit verification needs to happen after payload verification I think the two can happen in parallel. We'll have to pay close attention to how these invalid blob scenarios are described in gossip because only blobs passing all gossip conditions should be put in this cache:

https://github.com/ethereum/consensus-specs/pull/3244/files#diff-21cd1dde2a4815e714ef42d0cefa1bbd2997a4358ecf6e9e51025332c1d642fdR115-R116

The described scenarios would require the proposer to sign over invalid data so it's very likely this block gets orphaned.. but this is the type of thing we'll need to think very carefully about with regard to fork choice. Within a slot attesters should not be voting on unavailable block/blobs but we should have the ability to re-org to blocks that were not immediately available but became available to us and the rest of the network had available. (like if we later complete a by roots request for blobs)

Generally the halting criteria is the criteria for the completion of the future in the availability pending block. And this will have to be constrained by a timeout because at some point in block processing we will need to consider the data "unavailable". But this doesn't necessarily mean we should get rid of the block + blob for the above reasons. A failure of some types may mean we have to queue for reprocessing (waiting on the last blob).

[emhane]

My branch is starting to get to a good state now. As you know I scratched the shared-memory approach above of a separate blob and block cache, because the most sensible time point to send blobs to its block turned out to be as they came over the network, so it makes sense to have blobs and block in the same place. That however concerned me since the start, having many worker threads waiting on a write lock for the same cache, block workers and blob workers. So now I have built a pure* message-passing model around your

type DataAvailabilityHandle<E> = JoinHandle<Result<Option<Arc<BlobsSidecar<E>>>, BlobError>>; 

or more exactly with the new sidecar type

type DataAvailabilityHandle<E> = JoinHandle<Result<Option<Arc<SignedBlobSidecar<E>>>, BlobError>>; 

sending blobs to the block directly instead (or rather to a receiver that the block will pick up when it arrives).
(*It's not a completely pure message passing model, there is one global registry: a cache for the channel between a block and blob workers, but a block only accesses that once upon arrival.)

Basically, it works around three types and an enum implementing one trait TryIntoAvailableBlock:
Arc<SignedBeaconBlock>
AvailabilityPendingBlock
AvailableBlock
and an enum wrapping all three for when they can't be passed as generic (for example in ReprocessQueue as blocks may be in different stages of availability as they are in the queue...I don't know if they are, you tell me, but my implementation allows for that to happen)
SomeAvailabilityBlock.

At many points in the code however, we know the exact type of the three. For example all blocks that come over the network and are sent over the network are always of the type Arc<SignedBeaconBlock>. And all blocks that will be written to fork choice will be of type AvailableBlock. A block won't leave processing until it has transitioned from an AvailabilityPendingBlock to an AvailableBlock. So for many functions, composite types, trait implementations, etc. we actually know at time of coding what type to set the generic B: TryIntoAvailableBlock to. The way I implemented this, you guys can decide later where the most sensible place in the code is for a block to transition from an Arc<SignedBeaconBlock> into an AvailabilityPendingBlock.

There is one constraint with this message-passing model for decoupling that we have to work around (spelling availability correct many times in a row). An AvailabilityPendingBlock can't be cloned. This is due to its DataAvailabilityHandle. I have however found that there is no need to ever do so. Any type of batch processing that LH has been doing earlier relying on cloning an iterator over blocks can still be done by cloning the block inside the AvailabilityPendingBlock.

pub struct AvailabilityPendingBlock<E: EthSpec> {
    block: Arc<SignedBeaconBlock<E>>,
    data_availability_handle: DataAvailabilityHandle<E>,
}

Apart from that, a block in processing has one path that it travels that branches in Yes/No branches, there is no need to clone it. Say for example a block's parent is unknown, a check which LH calls at many different places in processing, some before and some after we try_into_available_block to import the block to fork choice, if the parent is unknown the block goes one way, if it is known, it goes another. It can go that way, and take its DataAvailabilityHandle with it.

A very important aspect of decoupling using the message-passing model becomes error handling. There are two new errors that are important to handle and not propagate DataAvailabilityFailure and BlobError::PendingAvailability. The first wraps the pieces, block and blobs, that an AvailabilityPendingBlock has gathered so far (if the block never comes, LH already has a mechanism for a block to be looked up via rpc as it will bother its children blocks and cause the ParentUnknown error, and the receiver for its blobs, with the blobs ready to receive on it for the case that they already came, will be there waiting for it in a pending_blocks_tx_rx entry for when it finally arrives). The PendingAvailability error is returned if try_into_available_block is called on a type B: TryIntoAvailableBlock that is not yet available. The error variant wraps an AvailabilityPendingBlock. As all AvailabilityPendingBlocks, this block is "active", it's a destination for blobs arriving on the network, we don't want to lose it by any other means than than it throwing an error, otherwise we will forever lose the blobs that have already arrived for the block, as well as not being able to make a new AvailabilityPendingBlock (the DataAvailabilityFailure error cleans up the pending_blocks_tx_rx removing the channel that failed block used).

The values I set for constants DEFAULT_DATA_AVAILABILITY_TIMEOUT and DEFAULT_PENDING_AVAILABILITY_BLOCKS are placeholders, someone with more experience of runtime benchmarks should set those.

There is a background thread I added for polling an AvailabilityPendingExecutedBlock (behind the scenes it polls the DataAvailabilityHandle on an AvailabilityPendingBlock). This is because, there is one place in the code that a block HAS to be available, and if it is not, we need a place to put it. There is no more processing path for it to travel down, until it is an AvailableBlock. This place is here: https://github.com/emhane/lighthouse/blob/84a191b0f03670270f76200670b08647c7d7b9e3/beacon_node/beacon_chain/src/beacon_chain.rs#L2884 . At this point, the block's payload has already been verified (that's an i/o op, don't want to wait for that again). So the block is wrapped up with its payload verification metadata and sent to the AvailabilityPendingCache. We might want to, on inserting the AvailabilityPendingExecutedBlock into the cache, notify the reprocessing queue that "hey! we have your parent!" so it isn't looked up over rpc for no reason, even though it would not be of harm as duplicate AvailabilityPendingBlocks are blocked with a BlobError::Duplicate.