-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
8 changed files
with
174 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| use crate::last_witness_cache_updater::MessageReceiver; | ||
| use chrono::{DateTime, Utc}; | ||
| use helium_crypto::PublicKeyBinary; | ||
|
|
||
| #[derive(Clone)] | ||
| pub struct LastWitnessCache { | ||
| last_witness_cache_receiver: MessageReceiver, | ||
| } | ||
|
|
||
| impl LastWitnessCache { | ||
| pub fn new(last_witness_cache_receiver: MessageReceiver) -> Self { | ||
| Self { | ||
| last_witness_cache_receiver, | ||
| } | ||
| } | ||
|
|
||
| pub async fn resolve_last_witness_info( | ||
| &self, | ||
| address: &PublicKeyBinary, | ||
| ) -> anyhow::Result<Option<DateTime<Utc>>> { | ||
| match self.last_witness_cache_receiver.borrow().get(address) { | ||
| Some(hit) => { | ||
| metrics::increment_counter!("last_witness_cache_hit"); | ||
| Ok(Some(*hit)) | ||
| } | ||
| None => { | ||
| metrics::increment_counter!("last_witness_cache_miss"); | ||
| Ok(None) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| use crate::last_witness::LastWitness; | ||
| use chrono::{DateTime, Duration, Utc}; | ||
| use futures::{future::LocalBoxFuture, stream::StreamExt, TryFutureExt}; | ||
| use helium_crypto::PublicKeyBinary; | ||
|
|
||
| use sqlx::PgPool; | ||
| use std::collections::HashMap; | ||
| use task_manager::ManagedTask; | ||
| use tokio::sync::watch; | ||
| use tokio::time; | ||
|
|
||
| pub type LastWitnessMap = HashMap<PublicKeyBinary, DateTime<Utc>>; | ||
| pub type MessageSender = watch::Sender<LastWitnessMap>; | ||
| pub type MessageReceiver = watch::Receiver<LastWitnessMap>; | ||
|
|
||
| pub struct LastWitnessUpdater { | ||
| pool: PgPool, | ||
| refresh_interval: Duration, | ||
| sender: MessageSender, | ||
| } | ||
|
|
||
| impl ManagedTask for LastWitnessUpdater { | ||
| fn start_task( | ||
| self: Box<Self>, | ||
| shutdown: triggered::Listener, | ||
| ) -> LocalBoxFuture<'static, anyhow::Result<()>> { | ||
| let handle = tokio::spawn(self.run(shutdown)); | ||
| Box::pin( | ||
| handle | ||
| .map_err(anyhow::Error::from) | ||
| .and_then(|result| async move { result.map_err(anyhow::Error::from) }), | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| impl LastWitnessUpdater { | ||
| pub async fn new( | ||
| pool: PgPool, | ||
| refresh_interval: Duration, | ||
| ) -> anyhow::Result<(MessageReceiver, Self)> { | ||
| let last_witness_map = refresh_last_witnesses(&pool).await?; | ||
| let (sender, receiver) = watch::channel(last_witness_map); | ||
| Ok(( | ||
| receiver, | ||
| Self { | ||
| pool, | ||
| refresh_interval, | ||
| sender, | ||
| }, | ||
| )) | ||
| } | ||
|
|
||
| pub async fn run(mut self, shutdown: triggered::Listener) -> anyhow::Result<()> { | ||
| tracing::info!("starting last_witness_updater"); | ||
| let mut trigger_timer = time::interval( | ||
| self.refresh_interval | ||
| .to_std() | ||
| .expect("valid interval in seconds for last witness cache updater"), | ||
| ); | ||
| loop { | ||
| tokio::select! { | ||
| biased; | ||
| _ = shutdown.clone() => break, | ||
| _ = trigger_timer.tick() => self.handle_refresh_tick().await?, | ||
| } | ||
| } | ||
| tracing::info!("stopping last_witness_updater"); | ||
| Ok(()) | ||
| } | ||
|
|
||
| async fn handle_refresh_tick(&mut self) -> anyhow::Result<()> { | ||
| tracing::info!("handling refresh tick"); | ||
| let updated_map = refresh_last_witnesses(&self.pool).await?; | ||
| let count = updated_map.len(); | ||
| if count > 0 { | ||
| tracing::info!("completed refreshing last witnesses, total count: {count}"); | ||
| self.sender.send(updated_map)?; | ||
| } else { | ||
| tracing::warn!("failed to refresh last witnesses, empty map..."); | ||
| } | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| pub async fn refresh_last_witnesses(pool: &PgPool) -> anyhow::Result<LastWitnessMap> { | ||
| tracing::info!("refreshing last witness cache"); | ||
| let mut map = LastWitnessMap::new(); | ||
| let mut stream = LastWitness::get_all(pool).await; | ||
| while let Some(last_witness_info) = stream.next().await.transpose()? { | ||
| map.insert(last_witness_info.id.into(), last_witness_info.timestamp); | ||
| } | ||
| Ok(map) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters