liquidator: forcefully exit process if snapshot job die (#924)
* liquidator: forcefully exit process if snapshot job die
* client: return snapshot_job join handle so it can be watched for early unexpected exit
(cherry picked from commit 2520c7d095
)
This commit is contained in:
parent
b55e900855
commit
bf6d2a11ec
|
@ -66,7 +66,7 @@ pub async fn save_snapshot(
|
|||
.await?;
|
||||
|
||||
// Getting solana account snapshots via jsonrpc
|
||||
snapshot_source::start(
|
||||
let snapshot_job = snapshot_source::start(
|
||||
snapshot_source::Config {
|
||||
rpc_http_url: rpc_url.clone(),
|
||||
mango_group,
|
||||
|
@ -78,6 +78,11 @@ pub async fn save_snapshot(
|
|||
oracles_and_vaults,
|
||||
account_update_sender,
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let res = snapshot_job.await;
|
||||
tracing::error!("Snapshot job exited, terminating process.. ({:?})", res);
|
||||
std::process::exit(-1);
|
||||
});
|
||||
|
||||
let mut chain_data = chain_data::ChainData::new();
|
||||
|
||||
|
|
|
@ -155,7 +155,7 @@ async fn main() -> anyhow::Result<()> {
|
|||
|
||||
// Getting solana account snapshots via jsonrpc
|
||||
// FUTURE: of what to fetch a snapshot - should probably take as an input
|
||||
snapshot_source::start(
|
||||
let snapshot_job = snapshot_source::start(
|
||||
snapshot_source::Config {
|
||||
rpc_http_url: rpc_url.clone(),
|
||||
mango_group,
|
||||
|
@ -457,6 +457,7 @@ async fn main() -> anyhow::Result<()> {
|
|||
liquidation_job,
|
||||
token_swap_info_job,
|
||||
check_changes_for_abort_job,
|
||||
snapshot_job,
|
||||
]
|
||||
.into_iter()
|
||||
.chain(prio_jobs.into_iter())
|
||||
|
|
|
@ -64,7 +64,8 @@ async fn main() -> anyhow::Result<()> {
|
|||
)
|
||||
.await?;
|
||||
|
||||
let mut jobs = vec![exit_processor.job, data_processor.job, health_processor.job];
|
||||
let mut jobs = vec![exit_processor.job, health_processor.job];
|
||||
jobs.extend(data_processor.jobs);
|
||||
|
||||
if let Some(logger) = logger {
|
||||
jobs.push(logger.job)
|
||||
|
|
|
@ -22,7 +22,7 @@ use tracing::warn;
|
|||
|
||||
pub struct DataProcessor {
|
||||
pub channel: tokio::sync::broadcast::Sender<DataEvent>,
|
||||
pub job: JoinHandle<()>,
|
||||
pub jobs: Vec<JoinHandle<()>>,
|
||||
pub chain_data: Arc<RwLock<chain_data::ChainData>>,
|
||||
}
|
||||
|
||||
|
@ -52,7 +52,7 @@ impl DataProcessor {
|
|||
) -> anyhow::Result<DataProcessor> {
|
||||
let mut retry_counter = RetryCounter::new(2);
|
||||
let mango_group = Pubkey::from_str(&configuration.mango_group)?;
|
||||
let mango_stream =
|
||||
let (mango_stream, snapshot_job) =
|
||||
fail_or_retry!(retry_counter, Self::init_mango_source(configuration).await)?;
|
||||
let (sender, _) = tokio::sync::broadcast::channel(8192);
|
||||
let sender_clone = sender.clone();
|
||||
|
@ -98,7 +98,7 @@ impl DataProcessor {
|
|||
|
||||
let result = DataProcessor {
|
||||
channel: sender,
|
||||
job,
|
||||
jobs: vec![job, snapshot_job],
|
||||
chain_data,
|
||||
};
|
||||
|
||||
|
@ -147,7 +147,9 @@ impl DataProcessor {
|
|||
return Some(Other);
|
||||
}
|
||||
|
||||
async fn init_mango_source(configuration: &Configuration) -> anyhow::Result<Receiver<Message>> {
|
||||
async fn init_mango_source(
|
||||
configuration: &Configuration,
|
||||
) -> anyhow::Result<(Receiver<Message>, JoinHandle<()>)> {
|
||||
//
|
||||
// Client setup
|
||||
//
|
||||
|
@ -192,7 +194,7 @@ impl DataProcessor {
|
|||
|
||||
// Getting solana account snapshots via jsonrpc
|
||||
// FUTURE: of what to fetch a snapshot - should probably take as an input
|
||||
snapshot_source::start(
|
||||
let snapshot_job = snapshot_source::start(
|
||||
snapshot_source::Config {
|
||||
rpc_http_url: configuration.rpc_http_url.clone(),
|
||||
mango_group,
|
||||
|
@ -205,6 +207,6 @@ impl DataProcessor {
|
|||
account_update_sender,
|
||||
);
|
||||
|
||||
Ok(account_update_receiver)
|
||||
Ok((account_update_receiver, snapshot_job))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -178,7 +178,7 @@ async fn main() -> anyhow::Result<()> {
|
|||
|
||||
// Getting solana account snapshots via jsonrpc
|
||||
// FUTURE: of what to fetch a snapshot - should probably take as an input
|
||||
snapshot_source::start(
|
||||
let snapshot_job = snapshot_source::start(
|
||||
snapshot_source::Config {
|
||||
rpc_http_url: rpc_url.clone(),
|
||||
mango_group,
|
||||
|
@ -353,6 +353,7 @@ async fn main() -> anyhow::Result<()> {
|
|||
|
||||
use futures::StreamExt;
|
||||
let mut jobs: futures::stream::FuturesUnordered<_> = vec![
|
||||
snapshot_job,
|
||||
data_job,
|
||||
settle_job,
|
||||
tcs_start_job,
|
||||
|
|
|
@ -16,6 +16,7 @@ use solana_rpc::rpc::rpc_accounts::AccountsDataClient;
|
|||
use solana_rpc::rpc::rpc_accounts_scan::AccountsScanClient;
|
||||
use std::str::FromStr;
|
||||
use std::time::Duration;
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::time;
|
||||
use tracing::*;
|
||||
|
||||
|
@ -221,11 +222,15 @@ async fn feed_snapshots(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
pub fn start(config: Config, mango_oracles: Vec<Pubkey>, sender: async_channel::Sender<Message>) {
|
||||
pub fn start(
|
||||
config: Config,
|
||||
mango_oracles: Vec<Pubkey>,
|
||||
sender: async_channel::Sender<Message>,
|
||||
) -> JoinHandle<()> {
|
||||
let mut poll_wait_first_snapshot = crate::delay_interval(time::Duration::from_secs(2));
|
||||
let mut interval_between_snapshots = crate::delay_interval(config.snapshot_interval);
|
||||
|
||||
tokio::spawn(async move {
|
||||
let snapshot_job = tokio::spawn(async move {
|
||||
let rpc_client = http::connect_with_options::<MinimalClient>(&config.rpc_http_url, true)
|
||||
.await
|
||||
.expect("always Ok");
|
||||
|
@ -258,4 +263,6 @@ pub fn start(config: Config, mango_oracles: Vec<Pubkey>, sender: async_channel::
|
|||
};
|
||||
}
|
||||
});
|
||||
|
||||
snapshot_job
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue