use crate::{EquivocationDetectionPipeline, SourceClient};
use futures::FutureExt;
use relay_utils::{TrackedTransactionFuture, TrackedTransactionStatus, TransactionTracker};
use std::{
future::poll_fn,
task::{Context, Poll},
};
pub struct EquivocationsReporter<'a, P: EquivocationDetectionPipeline, SC: SourceClient<P>> {
pending_reports: Vec<TrackedTransactionFuture<'a, SC::TransactionTracker>>,
}
impl<'a, P: EquivocationDetectionPipeline, SC: SourceClient<P>> EquivocationsReporter<'a, P, SC> {
pub fn new() -> Self {
Self { pending_reports: vec![] }
}
pub async fn submit_report(
&mut self,
source_client: &SC,
at: P::Hash,
equivocation: P::EquivocationProof,
) -> Result<(), SC::Error> {
let pending_report = source_client.report_equivocation(at, equivocation).await?;
self.pending_reports.push(pending_report.wait());
Ok(())
}
fn do_process_pending_reports(&mut self, cx: &mut Context<'_>) -> Poll<()> {
self.pending_reports.retain_mut(|pending_report| {
match pending_report.poll_unpin(cx) {
Poll::Ready(tx_status) => {
match tx_status {
TrackedTransactionStatus::Lost => {
log::error!(target: "bridge", "Equivocation report tx was lost");
},
TrackedTransactionStatus::Finalized(id) => {
log::error!(target: "bridge", "Equivocation report tx was finalized in source block {id:?}");
},
}
false
},
Poll::Pending => {
true
},
}
});
Poll::Ready(())
}
pub async fn process_pending_reports(&mut self) {
poll_fn(|cx| self.do_process_pending_reports(cx)).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::mock::*;
use relay_utils::HeaderId;
use std::sync::Mutex;
#[async_std::test]
async fn process_pending_reports_works() {
let polled_reports = Mutex::new(vec![]);
let finished_reports = Mutex::new(vec![]);
let mut reporter =
EquivocationsReporter::<TestEquivocationDetectionPipeline, TestSourceClient> {
pending_reports: vec![
Box::pin(async {
polled_reports.lock().unwrap().push(1);
finished_reports.lock().unwrap().push(1);
TrackedTransactionStatus::Finalized(HeaderId(1, 1))
}),
Box::pin(async {
polled_reports.lock().unwrap().push(2);
finished_reports.lock().unwrap().push(2);
TrackedTransactionStatus::Finalized(HeaderId(2, 2))
}),
Box::pin(async {
polled_reports.lock().unwrap().push(3);
std::future::pending::<()>().await;
finished_reports.lock().unwrap().push(3);
TrackedTransactionStatus::Finalized(HeaderId(3, 3))
}),
Box::pin(async {
polled_reports.lock().unwrap().push(4);
finished_reports.lock().unwrap().push(4);
TrackedTransactionStatus::Finalized(HeaderId(4, 4))
}),
],
};
reporter.process_pending_reports().await;
assert_eq!(*polled_reports.lock().unwrap(), vec![1, 2, 3, 4]);
assert_eq!(*finished_reports.lock().unwrap(), vec![1, 2, 4]);
assert_eq!(reporter.pending_reports.len(), 1);
}
}