From b9dcf220165d148a76b921e849cfcd017cc3f68f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ond=C5=99ej=20Hru=C5=A1ka?= Date: Sat, 18 Sep 2021 14:40:51 +0200 Subject: [PATCH] better reconnect and missed status handling --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/group_handler/mod.rs | 236 +++++++++++++++++++++++++++++---------- src/main.rs | 7 +- 4 files changed, 183 insertions(+), 64 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 801a0ac..e1a6059 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -328,7 +328,7 @@ checksum = "e88a8acf291dafb59c2d96e8f59828f3838bb1a70398823ade51a84de6a6deed" [[package]] name = "fedigroups" -version = "0.2.7" +version = "0.2.8" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index b9c1c96..e57ae17 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fedigroups" -version = "0.2.7" +version = "0.2.8" authors = ["Ondřej Hruška "] edition = "2018" publish = false diff --git a/src/group_handler/mod.rs b/src/group_handler/mod.rs index aaa7b7f..cd10a11 100644 --- a/src/group_handler/mod.rs +++ b/src/group_handler/mod.rs @@ -33,7 +33,7 @@ pub struct GroupHandle { // const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250); const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(500); -const MAX_CATCHUP_NOTIFS: usize = 25; +const MAX_CATCHUP_NOTIFS: usize = 30; // also statuses const MAX_CATCHUP_STATUSES: usize = 50; // higher because we can expect a lot of non-hashtag statuses here @@ -42,11 +42,60 @@ const SOCKET_ALIVE_TIMEOUT: Duration = Duration::from_secs(30); const SOCKET_RETIRE_TIME: Duration = Duration::from_secs(120); const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save! + + +macro_rules! grp_debug { + ($self:ident, $f:expr) => { + ::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + +macro_rules! grp_info { + ($self:ident, $f:expr) => { + ::log::info!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::info!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + +macro_rules! grp_trace { + ($self:ident, $f:expr) => { + ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + +macro_rules! grp_warn { + ($self:ident, $f:expr) => { + ::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + +macro_rules! grp_error { + ($self:ident, $f:expr) => { + ::log::error!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::error!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + + + impl GroupHandle { pub async fn save(&mut self) -> Result<(), GroupError> { - debug!("Saving group config & status"); + grp_debug!(self, "Saving group config & status"); self.store.set_group_config(self.config.clone()).await?; - trace!("Saved"); + grp_trace!(self, "Saved"); self.config.clear_dirty_status(); Ok(()) } @@ -88,39 +137,74 @@ impl NotifTimestamp for Status { } } +macro_rules! grp_trace { + ($self:ident, $f:expr) => { + ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct()); + }; + ($self:ident, $f:expr, $($arg:tt)+) => { + ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+); + }; +} + impl GroupHandle { pub async fn run(&mut self) -> Result<(), GroupError> { + loop { + match self.run_internal().await { + Ok(()) => unreachable!(), + Err(e @ GroupError::BadConfig(_)) => { + grp_error!(self, "ERROR in group handler, aborting! {}", e); + return Err(e); + } + Err(other) => { + grp_error!(self, "ERROR in group handler, will restart! {}", other); + tokio::time::sleep(Duration::from_secs(5)).await; + } + } + } + } + + pub async fn run_internal(&mut self) -> Result<(), GroupError> { assert!(PERIODIC_SAVE >= PING_INTERVAL); loop { - debug!("Opening streaming API socket"); + grp_debug!(self, "Opening streaming API socket"); let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start - let mut events = self.client.streaming_user().await?; + + // wrapped in a timeout, this seems like the only place the group could hang + // (https://git.ondrovo.com/MightyPork/group-actor/issues/8) + let mut events = match tokio::time::timeout(Duration::from_secs(3), self.client.streaming_user()).await { + Ok(Ok(events)) => events, + Ok(Err(e)) => return Err(e.into()), + Err(_) => { + return Err(GroupError::ApiTimeout); + } + }; + let socket_open_time = Instant::now(); let mut last_rx = Instant::now(); // let mut last_ping = Instant::now(); match self.catch_up_with_missed_notifications().await { Ok(true) => { - debug!("Some missed notifs handled"); + grp_debug!(self, "Some missed notifs handled"); } Ok(false) => { - debug!("No notifs missed"); + grp_debug!(self, "No notifs missed"); } Err(e) => { - error!("Failed to handle missed notifs: {}", e); + grp_error!(self, "Failed to handle missed notifs: {}", e); } } match self.catch_up_with_missed_statuses().await { Ok(true) => { - debug!("Some missed statuses handled"); + grp_debug!(self, "Some missed statuses handled"); } Ok(false) => { - debug!("No statuses missed"); + grp_debug!(self, "No statuses missed"); } Err(e) => { - error!("Failed to handle missed statuses: {}", e); + grp_error!(self, "Failed to handle missed statuses: {}", e); } } @@ -130,8 +214,10 @@ impl GroupHandle { } 'rx: loop { + // TODO add timeout to catch lock-ups! + if next_save < Instant::now() { - trace!("Save time elapsed, saving if needed"); + grp_trace!(self, "Save time elapsed, saving if needed"); self.save_if_needed().await.log_error("Failed to save group"); next_save = Instant::now() + PERIODIC_SAVE; } @@ -140,15 +226,15 @@ impl GroupHandle { let remains_to_retire = SOCKET_RETIRE_TIME.saturating_sub(socket_open_time.elapsed()); if remains_to_idle_close.is_zero() { - warn!("Socket idle too long, close"); + grp_warn!(self, "Socket idle too long, close"); break 'rx; } if remains_to_retire.is_zero() { - debug!("Socket open too long, closing"); + grp_debug!(self, "Socket open too long, closing"); break 'rx; } - trace!("Waiting for message"); + grp_trace!(self, "Waiting for message"); let timeout = next_save .saturating_duration_since(Instant::now()) .min(remains_to_idle_close) @@ -158,7 +244,7 @@ impl GroupHandle { match tokio::time::timeout(timeout, events.next()).await { Ok(Some(event)) => { last_rx = Instant::now(); - debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event)); + grp_debug!(self, "(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event)); match event { Event::Update(status) => { self.handle_status(status).await.log_error("Error handling a status"); @@ -172,7 +258,7 @@ impl GroupHandle { } } Ok(None) => { - warn!("Group @{} socket closed, restarting...", self.config.get_acct()); + grp_warn!(self, "Group @{} socket closed, restarting...", self.config.get_acct()); break 'rx; } Err(_) => { @@ -185,7 +271,7 @@ impl GroupHandle { // if last_ping.elapsed() > PING_INTERVAL { // last_ping = Instant::now(); - // trace!("Pinging"); + // grp_trace!(self, "Pinging"); // if events.send_ping() // .await.is_err() { // break 'rx; @@ -193,13 +279,13 @@ impl GroupHandle { // } } - warn!("Notif stream closed, will reopen"); + grp_warn!(self, "Notif stream closed, will reopen"); tokio::time::sleep(DELAY_REOPEN_STREAM).await; } } async fn handle_notification(&mut self, n: Notification) -> Result<(), GroupError> { - debug!("Handling notif #{}", n.id); + grp_debug!(self, "Handling notif #{}", n.id); let ts = n.timestamp_millis(); self.config.set_last_notif(ts); @@ -208,12 +294,12 @@ impl GroupHandle { let notif_acct = normalize_acct(&n.account.acct, &group_acct)?; if notif_acct == group_acct { - debug!("This is our post, ignore that"); + grp_debug!(self, "This is our post, ignore that"); return Ok(()); } if self.config.is_banned(¬if_acct) { - warn!("Notification actor {} is banned!", notif_acct); + grp_warn!(self, "Notification actor {} is banned!", notif_acct); return Ok(()); } @@ -224,14 +310,14 @@ impl GroupHandle { } } NotificationType::Follow => { - info!("New follower!"); + grp_info!(self, "New follower!"); // Just greet the user always self.handle_new_follow(¬if_acct, notif_user_id).await; // if self.config.is_member_or_admin(¬if_acct) { // // Already joined, just doing something silly, ignore this - // debug!("User already a member, ignoring"); + // grp_debug!(self, "User already a member, ignoring"); // } else { // // } @@ -246,68 +332,97 @@ impl GroupHandle { /// Handle a non-mention status for tags async fn handle_status(&mut self, s: Status) -> Result<(), GroupError> { - debug!("Handling status #{}", s.id); + grp_debug!(self, "Handling status #{}", s.id); let ts = s.timestamp_millis(); self.config.set_last_status(ts); - // Short circuit checks - if s.visibility.is_private() { - debug!("Status is direct/private, discard"); - return Ok(()); - } - - if !s.content.contains('#') { - debug!("No tags in status, discard"); - return Ok(()); - } - + let private = s.visibility.is_private(); + let has_hashtags = s.content.contains('#'); let group_user = self.config.get_acct(); let status_user = normalize_acct(&s.account.acct, group_user)?; + let member_or_admin = self.config.is_member_or_admin(&status_user); + let commands = crate::command::parse_slash_commands(&s.content); if status_user == group_user { - debug!("This is our post, discard"); + grp_debug!(self, "This is our post, discard"); return Ok(()); } if self.config.is_banned(&status_user) { - debug!("Status author @{} is banned, discard", status_user); + grp_debug!(self, "Status author @{} is banned, discard", status_user); return Ok(()); } - // optout does not work for members and admins, so don't check it - - if !self.config.is_member_or_admin(&status_user) { - debug!("Status author @{} is not a member, discard", status_user); + if self.config.is_optout(&status_user) && !member_or_admin + { + grp_debug!(self, "Status author @{} opted out, discard", status_user); return Ok(()); } - let commands = crate::command::parse_slash_commands(&s.content); if commands.contains(&StatusCommand::Ignore) { - debug!("Post has IGNORE command, discard"); + grp_debug!(self, "Post has IGNORE command, discard"); return Ok(()); } - for m in s.mentions { + // Sometimes notifications don't work, but we see the mentions as statuses + for m in &s.mentions { let mentioned_user = normalize_acct(&m.acct, group_user)?; if mentioned_user == group_user { - if !commands.is_empty() { - debug!("Detected commands for this group, tags dont apply; discard"); + let notif_time = self.config.get_last_notif(); + if notif_time <= ts { + grp_debug!(self, "mentioned but status is older than last notif, can't be a valid notif, discard"); return Ok(()); + } else { + if !commands.is_empty() { + grp_debug!(self, "Detected commands for this group, handle as notif"); + + return self.handle_notification(Notification { + id: s.id.clone(), // ??? + notification_type: NotificationType::Mention, + created_at: s.created_at.clone(), + account: s.account.clone(), + status: Some(s) + }).await; + } else { + if !private { + grp_debug!(self, "Detected mention status, handle as notif"); + } else { + grp_debug!(self, "mention in private without commands, discard, this is nothing"); + return Ok(()); + } + } } } } + // optout does not work for members and admins, so don't check it + + if !member_or_admin { + grp_debug!(self, "Status author @{} is not a member, discard", status_user); + return Ok(()); + } + + if private { + grp_debug!(self, "Status is private, discard"); + return Ok(()); + } + + if !has_hashtags { + grp_debug!(self, "No hashtags, discard"); + return Ok(()); + } + let tags = crate::command::parse_status_tags(&s.content); - debug!("Tags in status: {:?}", tags); + grp_debug!(self, "Tags in status: {:?}", tags); 'tags: for t in tags { if self.config.is_tag_followed(&t) { - info!("REBLOG #{} STATUS", t); + grp_info!(self, "REBLOG #{} STATUS", t); self.client.reblog(&s.id).await .log_error("Failed to reblog"); break 'tags; // do not reblog multiple times! } else { - debug!("#{} is not a group tag", t); + grp_debug!(self, "#{} is not a group tag", t); } } @@ -337,11 +452,11 @@ impl GroupHandle { break; // reached our last seen notif } - debug!("Inspecting notif {}", NotificationDisplay(&n)); + grp_debug!(self, "Inspecting notif {}", NotificationDisplay(&n)); notifs_to_handle.push(n); num += 1; if num > MAX_CATCHUP_NOTIFS { - warn!("Too many notifs missed to catch up!"); + grp_warn!(self, "Too many notifs missed to catch up!"); break; } @@ -355,10 +470,10 @@ impl GroupHandle { notifs_to_handle.reverse(); - debug!("{} notifications to catch up!", notifs_to_handle.len()); + grp_debug!(self, "{} notifications to catch up!", notifs_to_handle.len()); for n in notifs_to_handle { - debug!("Handling missed notification: {}", NotificationDisplay(&n)); + grp_debug!(self, "Handling missed notification: {}", NotificationDisplay(&n)); self.handle_notification(n).await.log_error("Error handling a notification"); } @@ -385,18 +500,16 @@ impl GroupHandle { break; // reached our last seen status (hopefully there arent any retro-bumped) } - debug!("Inspecting status {}", StatusDisplay(&s)); + grp_debug!(self, "Inspecting status {}", StatusDisplay(&s)); if newest_status.is_none() { newest_status = Some(ts); } - if s.content.contains('#') && !s.visibility.is_private() { - statuses_to_handle.push(s); - } + statuses_to_handle.push(s); num += 1; if num > MAX_CATCHUP_STATUSES { - warn!("Too many statuses missed to catch up!"); + grp_warn!(self, "Too many statuses missed to catch up!"); break; } @@ -409,15 +522,16 @@ impl GroupHandle { } if statuses_to_handle.is_empty() { + grp_debug!(self, "No statuses to handle"); return Ok(false); } statuses_to_handle.reverse(); - debug!("{} statuses to catch up!", statuses_to_handle.len()); + grp_debug!(self, "{} statuses to catch up!", statuses_to_handle.len()); for s in statuses_to_handle { - debug!("Handling missed status: {}", StatusDisplay(&s)); + grp_debug!(self, "Handling missed status: {}", StatusDisplay(&s)); self.handle_status(s).await .log_error("Error handling a status"); } diff --git a/src/main.rs b/src/main.rs index ce7121d..8244d40 100644 --- a/src/main.rs +++ b/src/main.rs @@ -116,7 +116,12 @@ async fn main() -> anyhow::Result<()> { let mut handles = vec![]; for mut g in groups { - handles.push(tokio::spawn(async move { g.run().await })); + handles.push(tokio::spawn(async move { + match g.run().await { + Ok(()) => unreachable!(), + Err(e) => error!("GROUP FAILED! {}", e), + } + })); } futures::future::join_all(handles).await;