uqbar-dao/nectar
1
1
Fork 0
mirror of https://github.com/uqbar-dao/nectar.git synced 2024-12-23 16:43:24 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: kill node<>node eth subs after 2 hours no data

Browse Source
This commit is contained in:
dr-frmr 2024-05-08 13:24:43 -06:00
parent b79c9535df
commit 6a400dd850
No known key found for this signature in database
1 changed files with 15 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
kinode/src/eth/subscription.rs
View File

@ -300,6 +300,7 @@ async fn build_subscription(
} }
} }
} }
response_channels.remove(&km_id);
return Err(EthError::NoRpcForChain); return Err(EthError::NoRpcForChain);
} }
@ -338,6 +339,9 @@ async fn maintain_local_subscription(
/// handle the subscription updates from a remote provider, /// handle the subscription updates from a remote provider,
/// and also perform keepalive checks on that provider. /// and also perform keepalive checks on that provider.
/// current keepalive is 30s, this can be adjusted as desired /// current keepalive is 30s, this can be adjusted as desired
///
/// if the subscription goes more than 2 hours without an update,
/// the provider will be considered dead and the subscription will be closed.
async fn maintain_remote_subscription( async fn maintain_remote_subscription(
our: &str, our: &str,
provider_node: &str, provider_node: &str,
@ -350,11 +354,16 @@ async fn maintain_remote_subscription(
send_to_loop: &MessageSender, send_to_loop: &MessageSender,
) -> Result<(), EthSubError> { ) -> Result<(), EthSubError> {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30)); let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
let mut last_received = tokio::time::Instant::now();
let two_hours = tokio::time::Duration::from_secs(2 * 3600);
loop { loop {
tokio::select! { tokio::select! {
incoming = rx.recv() => { incoming = rx.recv() => {
match incoming { match incoming {
Some(EthSubResult::Ok(upd)) => { Some(EthSubResult::Ok(upd)) => {
// Update the last received time on any successful sub result
last_received = tokio::time::Instant::now();
kernel_message( kernel_message(
&our, &our,
rand::random(), rand::random(),
@ -381,7 +390,6 @@ async fn maintain_remote_subscription(
id: sub_id, id: sub_id,
error: "subscription closed unexpectedly".to_string(), error: "subscription closed unexpectedly".to_string(),
}); });
} }
} }
} }
@ -406,6 +414,12 @@ async fn maintain_remote_subscription(
}); });
} }
} }
_ = tokio::time::sleep_until(last_received + two_hours) => {
return Err(EthSubError {
id: sub_id,
error: "No updates received for 2 hours, subscription considered dead.".to_string(),
});
}
} }
} }
} }