Support for reorged logs in eth_Logs Subscription #5851
Conversation
Arindam2407
requested review from
mattsse,
Rjected,
Evalir,
onbjerg and
gakonst
as code owners
Arindam2407
changed the title
crates/rpc/rpc/src/eth/filter.rs
Outdated
| /// Reacts to reorged blocks, checks impacted log filters, stores reorged logs in the log filter | ||
| pub async fn reorged_logs_stream( | ||
| &self, | ||
| pubsub: EthPubSub<Provider, Pool, Events, Network>, |
There was a problem hiding this comment.
I'm not exactly sure what this does.
this looks a bit convoluted.
how the on reorg logic should work is:
- listen for reorged blocks:
- acquire a lock on the shared state
- try to find matching filters, if so put the logs into the ActiveFilter
There was a problem hiding this comment.
- The temp_reorged_blocks stores consecutive removed blocks as they come. The matching logs for the given filter are stored in reverted_logs.
- If the hash of one of these blocks matches with the last polled hash of the filter, we know that the filter previously served all the blocks stored in temp_reorged_blocks. We can push the reverted logs in the reorged_logs vector along with the filter id.
- If the last polled hash does not match before the block subscription receives a newly included block, we can simply clear the temp_reorged_blocks.
Arindam2407
force-pushed
the
reorged_logs
branch
from
c3e4d38 to
8dff31e
Compare
Arindam2407
force-pushed
the
reorged_logs
branch
from
3fbd7d6 to
1cc3576
Compare
|
Please, could you review @mattsse ? Have made the requested changes |
crates/rpc/rpc/src/eth/filter.rs
Outdated
| provider: Provider, | ||
| pool: Pool, | ||
| eth_cache: EthStateCache, | ||
| config: EthFilterConfig, | ||
| task_spawner: Box<dyn TaskSpawner>, | ||
| ) -> Self { | ||
| pubsub: EthPubSub<Provider, Pool, Events, Network>, |
There was a problem hiding this comment.
we don't need this type, we only need events
There was a problem hiding this comment.
can you explain this further?
There was a problem hiding this comment.
we only need the events trait not the pubsub isntance
crates/rpc/rpc/src/eth/filter.rs
Outdated
| if removed { | ||
| temp_reorged_blocks.push(block_receipts); | ||
|
|
||
| let filters = executor::block_on(self.active_filters().inner.lock()); |
There was a problem hiding this comment.
no executor block,
this needs to be refactored so we can have async code
I suggest not do the flat_map stuff and instead move all of this to async code in the while loop
crates/rpc/rpc/src/eth/filter.rs
Outdated
|
|
||
| filters.iter().for_each(|(id, active_filter)| { | ||
| if let FilterKind::Log(ref filter) = active_filter.kind { | ||
| let mut reverted_logs: Vec<Log> = Vec::new(); |
There was a problem hiding this comment.
we need to check here first if the filter is even affected by this, by checking the block number and only check for removed logs if the filter is at the same height
crates/rpc/rpc/src/eth/filter.rs
Outdated
| let mut filters = executor::block_on(self.active_filters().inner.lock()); | ||
| let active_filter = | ||
| filters.get_mut(&id).ok_or(FilterError::FilterNotFound(id)).unwrap(); | ||
| let mut guard = active_filter.reorged_logs.lock().await; | ||
| if let Some(reorged_logs_vec) = guard.as_mut() { | ||
| reorged_logs_vec.append(&mut logs); | ||
| } |
There was a problem hiding this comment.
here we should handle the Reorg events, basically the flat map logic from below
crates/rpc/rpc/src/eth/filter.rs
Outdated
| /// Last time this filter was polled. | ||
| last_poll_timestamp: Instant, | ||
| /// What kind of filter it is. | ||
| kind: FilterKind, | ||
| /// Reorged logs | ||
| reorged_logs: Arc<Mutex<Option<Vec<Log>>>>, |
There was a problem hiding this comment.
since we already lock the active filters set, we don't need a mutex here, instead we can push removed logs in here directly when iterating over all active filters
Added improved implementation for reorged logs subscription
Arindam2407
force-pushed
the
reorged_logs
branch
from
7513fde to
51d367a
Compare
github-actions
bot
added
the
S-stale
Addressing issue #5677
Updated EthFilter to create new task update_reorged_logs that is spawned when the EthFilter is initialized. The new task reacts to reorged blocks taking a clone of EthPubSub, checks impacted log filters and stores the reorged logs in the log filter in a new field called reorged_logs.