better reconnect and missed status handling

Cargo.lock

@@ -328,7 +328,7 @@ checksum = "e88a8acf291dafb59c2d96e8f59828f3838bb1a70398823ade51a84de6a6deed"
 
 [[package]]
 name = "fedigroups"
-version = "0.2.7"
+version = "0.2.8"
 dependencies = [
  "anyhow",
  "clap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fedigroups"
-version = "0.2.7"
+version = "0.2.8"
 authors = ["Ondřej Hruška <ondra@ondrovo.com>"]
 edition = "2018"
 publish = false

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(&notif_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(&notif_acct, notif_user_id).await;
 
                 // if self.config.is_member_or_admin(&notif_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
+        let private = s.visibility.is_private();
-        if s.visibility.is_private() {
+        let has_hashtags = s.content.contains('#');
-            debug!("Status is direct/private, discard");
-            return Ok(());
-        }
-
-        if !s.content.contains('#') {
-            debug!("No tags in status, discard");
-            return Ok(());
-        }
-
         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(());
         }
 
+        if self.config.is_optout(&status_user) && !member_or_admin
+        {
+            grp_debug!(self, "Status author @{} opted out, discard", status_user);
+            return Ok(());
+        }
+
+        if commands.contains(&StatusCommand::Ignore) {
+            grp_debug!(self, "Post has IGNORE command, discard");
+            return Ok(());
+        }
+
+        // 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 {
+                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 !self.config.is_member_or_admin(&status_user) {
+        if !member_or_admin {
-            debug!("Status author @{} is not a member, discard", status_user);
+            grp_debug!(self, "Status author @{} is not a member, discard", status_user);
             return Ok(());
         }
 
-        let commands = crate::command::parse_slash_commands(&s.content);
+        if private {
-        if commands.contains(&StatusCommand::Ignore) {
+            grp_debug!(self, "Status is private, discard");
-            debug!("Post has IGNORE command, discard");
             return Ok(());
         }
 
-        for m in s.mentions {
+        if !has_hashtags {
-            let mentioned_user = normalize_acct(&m.acct, group_user)?;
+            grp_debug!(self, "No hashtags, discard");
-            if mentioned_user == group_user {
-                if !commands.is_empty() {
-                    debug!("Detected commands for this group, tags dont apply; 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);
-            }
             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");
         }

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;