Skip to content

feat: add reconnection logic to websockets #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
prenaissance merged 1 commit into from
Dec 15, 2025
Merged

feat: add reconnection logic to websockets #9

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
223 changes: 132 additions & 91 deletions src/processing/worker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,122 +12,163 @@ use super::pricing_event::{ListingType, PricingEvent};

const WS_URL: &str = "wss://ws.backpack.tf/events";
const PROCESSED_STATISTICS_INTERVAL_MINS: i64 = 5;
const MAX_BACKOFF_SECS: u64 = 300;

pub async fn run(
pool: AsyncDbPool,
cached_exchange_rate: CachedExchangeRate,
blocked_user_steam_ids: Arc<tokio::sync::Mutex<HashSet<String>>>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut processed_events_counter: i64 = 0;
) -> () {
let mut processed_events_counter = 0;
let mut last_statistics_log: Option<chrono::DateTime<chrono::Utc>> = None;
let (ws_stream, _response) = tokio_tungstenite::connect_async(WS_URL).await?;
tracing::info!("Connected to WS server");
let (_, mut read) = ws_stream.split();
while let Some(Ok(msg)) = read.next().await {
match msg {
Message::Text(text) => {
let events: Vec<PricingEvent> = match serde_json::from_str(&text) {
Ok(ev) => ev,
Err(e) => {
let file_name = format!(
"failed_ws_message_{}.json",
chrono::Utc::now().timestamp_millis()
);
if let Err(err) = tokio::fs::create_dir_all("/tmp/pricing_tf").await {
tracing::error!(
"Failed to create /tmp/pricing_tf directory to report deserialization errors. Continuing execution. Error: {}",
err
let mut backoff = 5;
loop {
let (ws_stream, _response) = match tokio_tungstenite::connect_async(WS_URL).await {
Ok(res) => res,
Err(e) => {
tracing::error!(
"Failed to connect to WS server: {}. Retrying in {} seconds...",
e,
backoff
);
tokio::time::sleep(std::time::Duration::from_secs(backoff)).await;
backoff = std::cmp::min(backoff * 2, MAX_BACKOFF_SECS);
continue;
}
};
tracing::info!("Connected to WS server");

let (_, mut read) = ws_stream.split();
while let Some(Ok(msg)) = read.next().await {
match msg {
Message::Text(text) => {
let events: Vec<PricingEvent> = match serde_json::from_str(&text) {
Ok(ev) => ev,
Err(e) => {
let file_name = format!(
"failed_ws_message_{}.json",
chrono::Utc::now().timestamp_millis()
);
continue;
}
if let Err(err) = tokio::fs::create_dir_all("/tmp/pricing_tf").await {
tracing::error!(
"Failed to create /tmp/pricing_tf directory to report deserialization errors. Continuing execution. Error: {}",
err
);
continue;
}

if let Err(err) = tokio::fs::write(
format!("/tmp/pricing_tf/{}", file_name),
text.as_bytes(),
)
.await
{
tracing::error!(
"Failed to write failed WS message to /tmp/pricing_tf/{}. Continuing execution. Error: {}",
file_name,
err
);
continue;
}

if let Err(err) = tokio::fs::write(
format!("/tmp/pricing_tf/{}", file_name),
text.as_bytes(),
)
.await
{
tracing::error!(
"Failed to write failed WS message to /tmp/pricing_tf/{}. Continuing execution. Error: {}",
file_name,
err
"Failed to deserialize message: {}\nJSON stored in /tmp/pricing_tf/{}",
e,
file_name
);

continue;
}
};
let steam_ids = blocked_user_steam_ids.lock().await;
let (upserts, deletes) = events
.into_iter()
// How does this even happen?
.filter(|event| !event.payload.item.name.is_empty())
.filter(|event| !steam_ids.contains(&event.payload.steamid))
.filter(|event| !etl::is_spelled_item(event))
.filter(|event| !etl::is_unusual_weapon(event))
.partition::<Vec<_>, _>(|event| event.event == ListingType::ListingUpdate);

tracing::error!(
"Failed to deserialize message: {}\nJSON stored in /tmp/pricing_tf/{}",
e,
file_name
);
let upserts = etl::filter_unique_listing_events(upserts);

continue;
}
};
let steam_ids = blocked_user_steam_ids.lock().await;
let (upserts, deletes) = events
.into_iter()
// How does this even happen?
.filter(|event| !event.payload.item.name.is_empty())
.filter(|event| !steam_ids.contains(&event.payload.steamid))
.filter(|event| !etl::is_spelled_item(event))
.filter(|event| !etl::is_unusual_weapon(event))
.partition::<Vec<_>, _>(|event| event.event == ListingType::ListingUpdate);
let upserts_len = upserts.len();
let deletes_len = deletes.len();

let upserts = etl::filter_unique_listing_events(upserts);
let mut connection = match pool.get().await {
Ok(conn) => conn,
Err(e) => {
tracing::error!(
"Failed to get DB connection from pool: {}. Skipping this batch of events.",
e
);
continue;
}
};

let upserts_len = upserts.len();
let deletes_len = deletes.len();
if !upserts.is_empty() {
let exchange_rate = cached_exchange_rate.lock().await.rate;

let mut connection = pool.get().await?;
let new_listings: Vec<_> = upserts
.into_iter()
.map(|event| event.to_trade_listing(exchange_rate))
.collect();

if !upserts.is_empty() {
let exchange_rate = cached_exchange_rate.lock().await.rate;
if let Err(e) =
etl::upsert_trade_listings(&mut connection, new_listings).await
{
tracing::error!(
"Failed to upsert trade listings into DB: {}. Skipping this batch of events.",
e
);
continue;
}
}
if !deletes.is_empty() {
let ids_to_delete: Vec<String> =
deletes.into_iter().map(|event| event.payload.id).collect();

let new_listings: Vec<_> = upserts
.into_iter()
.map(|event| event.to_trade_listing(exchange_rate))
.collect();
if let Err(e) =
etl::delete_trade_listings(&mut connection, ids_to_delete).await
{
tracing::error!(
"Failed to delete trade listings from DB: {}. Skipping this batch of events.",
e
);
continue;
};
}

etl::upsert_trade_listings(&mut connection, new_listings).await?;
}
if !deletes.is_empty() {
let ids_to_delete: Vec<String> =
deletes.into_iter().map(|event| event.payload.id).collect();
tracing::debug!(
"Successfully processed an event batch of {} upserts and {} deletes",
upserts_len,
deletes_len
);

etl::delete_trade_listings(&mut connection, ids_to_delete).await?;
processed_events_counter += (upserts_len + deletes_len) as i64;
let now = chrono::Utc::now();
if last_statistics_log.is_none_or(|last_statistics_log| {
now.signed_duration_since(last_statistics_log).num_minutes()
>= PROCESSED_STATISTICS_INTERVAL_MINS
}) {
tracing::info!(
"Processed {} events since the start of the program",
processed_events_counter
);
last_statistics_log = Some(now);
}
}

tracing::debug!(
"Successfully processed an event batch of {} upserts and {} deletes",
upserts_len,
deletes_len
);

processed_events_counter += (upserts_len + deletes_len) as i64;
let now = chrono::Utc::now();
if last_statistics_log.is_none_or(|last_statistics_log| {
now.signed_duration_since(last_statistics_log).num_minutes()
>= PROCESSED_STATISTICS_INTERVAL_MINS
}) {
tracing::info!(
"Processed {} events since the start of the program",
processed_events_counter
);
last_statistics_log = Some(now);
Message::Close(_) => {
tracing::warn!("WS Server closed connection.");
}
}
Message::Close(_) => {
tracing::warn!("WS Server closed connection.");
}
msg => {
if msg.is_empty() {
tracing::trace!("Received empty message");
continue;
msg => {
if msg.is_empty() {
tracing::trace!("Received empty message");
continue;
}
tracing::warn!("Unhandled message received: {}", msg);
}
tracing::warn!("Unhandled message received: {}", msg);
}
}
}
Ok(())
}