Processes updates from the queue until the channel closes.
(&self)
| 129 | |
| 130 | /// Processes updates from the queue until the channel closes. |
| 131 | pub async fn start(&self) { |
| 132 | while let Some(update) = self.rx.lock().await.recv().await { |
| 133 | let prev_pending_blocks = self.pending_blocks.load_full(); |
| 134 | match update { |
| 135 | StateUpdate::Canonical(block) => { |
| 136 | debug!(message = "processing canonical block", block_number = block.number); |
| 137 | match self.process_canonical_block(prev_pending_blocks, &block) { |
| 138 | Ok(new_pending_blocks) => { |
| 139 | self.pending_blocks.swap(new_pending_blocks); |
| 140 | |
| 141 | let mut cache = self.cache.lock().await; |
| 142 | cache.update_canonical(block.number); |
| 143 | let cached = cache.drain(block.number + 1); |
| 144 | drop(cache); |
| 145 | |
| 146 | if !cached.is_empty() { |
| 147 | debug!( |
| 148 | message = "replaying cached flashblocks after canonical block", |
| 149 | canonical_block = block.number, |
| 150 | cached_count = cached.len(), |
| 151 | ); |
| 152 | for flashblock in cached { |
| 153 | let fb_prev = self.pending_blocks.load_full(); |
| 154 | self.apply_flashblock(fb_prev, flashblock).await; |
| 155 | } |
| 156 | } |
| 157 | } |
| 158 | Err(e) => { |
| 159 | error!(message = "could not process canonical block", error = %e); |
| 160 | } |
| 161 | } |
| 162 | } |
| 163 | StateUpdate::Flashblock(flashblock) => { |
| 164 | debug!( |
| 165 | message = "processing flashblock", |
| 166 | block_number = flashblock.metadata.block_number, |
| 167 | flashblock_index = flashblock.index |
| 168 | ); |
| 169 | self.apply_flashblock(prev_pending_blocks, flashblock).await; |
| 170 | } |
| 171 | } |
| 172 | } |
| 173 | } |
| 174 | |
| 175 | async fn apply_flashblock( |
| 176 | &self, |
nothing calls this directly
no test coverage detected