Struct BlockSynchronizer

pub struct BlockSynchronizer<S> { /* private fields */ }

Aligns multiple StateSynchronizers on the block dimension.

Purpose

The purpose of this component is to handle streams from multiple state synchronizers and align/merge them according to their blocks. Ideally this should be done in a fault-tolerant way, meaning we can recover from a state synchronizer suffering from timing issues. E.g. a delayed or unresponsive state synchronizer might recover again, or an advanced state synchronizer can be included again once we reach the block it is at.

Limitations

• Supports only chains with fixed blocks time for now due to the lock step mechanism.

Initialisation

Queries all registered synchronizers for their first message and evaluates the state of each synchronizer. If a synchronizer’s first message is an older block, it is marked as delayed. If no message is received within the startup timeout, the synchronizer is marked as stale and is closed.

Main loop

Once started, the synchronizers are queried concurrently for messages in lock step: the main loop queries all synchronizers in ready for the last emitted data, builds the FeedMessage and emits it, then it schedules the wait procedure for the next block.

Synchronization Logic

To classify a synchronizer as delayed, we need to first define the current block. The highest block number of all ready synchronizers is considered the current block.

Once we have the current block we can easily determine which block we expect next. And if a synchronizer delivers an older block we can classify it as delayed.

If any synchronizer is not in the ready state we will try to bring it back to the ready state. This is done by trying to empty any buffers of a delayed synchronizer or waiting to reach the height of an advanced synchronizer (and flagging it as such in the meantime).

Of course, we can’t wait forever for a synchronizer to reply/recover. All of this must happen within the block production step of the blockchain: The wait procedure consists of waiting for any of the receivers to emit a new message (within a max timeout - several multiples of the block time). Once a message is received a very short timeout start for the remaining synchronizers, to deliver a message. Any synchronizer failing to do so is transitioned to delayed.

Note

The described process above is the goal. It is currently not implemented like that. Instead we simply wait block_time + wait_time. Synchronizers are expected to respond within that timeout. This is simpler but only works well on chains with fixed block times.

Implementations

impl<S> BlockSynchronizer<S>

where S: StateSynchronizer,

pub fn new( block_time: Duration, max_wait: Duration, max_missed_blocks: u64, ) -> Self

pub fn max_messages(&mut self, val: usize)

pub fn register_synchronizer( self, id: ExtractorIdentity, synchronizer: S, ) -> Self

pub async fn run( self, ) -> Result<(JoinHandle<()>, Receiver<FeedMessage>), BlockSynchronizerError>