improve robustness of recorder implementation

Author: dominikks

backend/src/discord/recorder.rs

@@ -17,6 +17,7 @@ use std::collections::HashMap;
 use std::collections::VecDeque;
 use std::env::var;
 use std::fmt;
+use std::num::Wrapping;
 use std::ops::Deref;
 use std::path::PathBuf;
 use std::process::Stdio;
@@ -25,6 +26,7 @@ use std::time::SystemTime;
 use tokio::fs;
 use tokio::io::AsyncWriteExt;
 use tokio::process::Command;
+use tokio::sync::MutexGuard;
 use tokio::time::sleep;
 use tokio::time::Duration;
 use tracing::Instrument;
@@ -78,6 +80,7 @@ struct VoiceRecording {
 struct UserData {
     user_id: UserId,
     last_voice_activity: SystemTime,
+    last_rtp_timestamp: Wrapping<u32>,
     recordings: VecDeque<VoiceRecording>,
 }
 
@@ -110,61 +113,27 @@ impl VoiceEventHandler for GuildRecorderArc {
                 if let Some(user_id) = user_id {
                     let is_new_user;
                     {
-                        let mut users = self.users.write().await;
+                        let users = self.users.read().await;
                         is_new_user = !users.contains_key(ssrc);
+                    }
+                    debug!(?is_new_user);
 
-                        if is_new_user {
+                    if is_new_user {
+                        {
+                            let mut users = self.users.write().await;
                             users.insert(
                                 *ssrc,
                                 Arc::new(Mutex::new(UserData {
                                     user_id: *user_id,
                                     last_voice_activity: SystemTime::now(),
+                                    last_rtp_timestamp: Wrapping(0),
                                     recordings: Default::default(),
                                 })),
                             );
                         }
-                    }
-                    debug!(?is_new_user);
 
-                    // When this user was newly added and did not exist before, we spawn a garbage collector
-                    if is_new_user {
-                        let users_lock = self.users.clone();
-                        let ssrc = *ssrc;
-                        let span = span!(Level::INFO, "user_gc", ssrc, ?user_id);
-                        tokio::spawn(
-                            async move {
-                                // Delete the user if no voice activity in the last hour
-                                loop {
-                                    trace!("Sleeping");
-                                    sleep(Duration::from_secs(60 * 60 * 2)).await;
-
-                                    let delete;
-                                    {
-                                        let users = users_lock.read().await;
-                                        if let Some(user_mutex) = users.get(&ssrc) {
-                                            let user = user_mutex.lock().await;
-                                            delete = SystemTime::now()
-                                                .duration_since(user.last_voice_activity)
-                                                .unwrap()
-                                                > Duration::from_secs(60 * 60);
-                                        } else {
-                                            debug!("User already deleted");
-                                            break;
-                                        }
-                                    }
-
-                                    debug!(delete, "Checking timeout of user data");
-
-                                    if delete {
-                                        let mut users = users_lock.write().await;
-                                        users.remove(&ssrc);
-                                        debug!("{} remaining known users", users.len());
-                                        break;
-                                    }
-                                }
-                            }
-                            .instrument(span),
-                        );
+                        // When this user was newly added and did not exist before, we spawn a garbage collector
+                        self.spawn_user_gc(*ssrc);
                     }
                 }
             }
@@ -175,38 +144,16 @@ impl VoiceEventHandler for GuildRecorderArc {
                     if *speaking { "started" } else { "stopped" },
                 );
 
-                let user_lock;
-                {
-                    let users = self.users.read().await;
-                    user_lock = users.get(ssrc).cloned()?;
-                }
+                // Once a user stops speaking, we cleanup their recordings
+                if !*speaking {
+                    let user_lock;
+                    {
+                        let users = self.users.read().await;
+                        user_lock = users.get(ssrc).cloned()?;
+                    }
 
-                if *speaking {
-                    // Create a new recording if the user starts speaking. We assume that all audio packets we receive between
-                    // a user starting and stopping to speak are adjunct.
                     let mut user = user_lock.lock().await;
-                    user.recordings.push_back(VoiceRecording {
-                        timestamp: SystemTime::now(),
-                        data: Vec::new(),
-                    });
-                } else {
-                    // Start a timeout for the recording. This starts when the user stops speaking.
-                    let span = span!(Level::INFO, "recording_gc", ssrc);
-                    tokio::spawn(
-                        async move {
-                            sleep(Duration::from_secs(*RECORDING_LENGTH)).await;
-
-                            let mut user = user_lock.lock().await;
-                            user.recordings
-                                .pop_front()
-                                .expect("Missing element in Deque");
-                            debug!(
-                                remaining_recordings = user.recordings.len(),
-                                "Removed timed out recording"
-                            );
-                        }
-                        .instrument(span),
-                    );
+                    self.cleanup_user_recordings(&mut user);
                 }
             }
             Ctx::VoicePacket(VoiceData { audio, packet, .. }) => {
@@ -218,7 +165,7 @@ impl VoiceEventHandler for GuildRecorderArc {
                         "Audio packet sequence {:05} has {:04} bytes (decompressed from {})",
                         packet.sequence.0,
                         audio.len() * std::mem::size_of::<i16>(),
-                        packet.payload.len()
+                        packet.payload.len(),
                     );
 
                     let user_lock;
@@ -227,12 +174,25 @@ impl VoiceEventHandler for GuildRecorderArc {
                         user_lock = users.get(&packet.ssrc).cloned()?;
                     }
 
-                    // Append the audio to the latest recording
                     let mut user = user_lock.lock().await;
-                    user.last_voice_activity = SystemTime::now();
-                    if let Some(recording) = user.recordings.back_mut() {
-                        recording.data.extend(audio);
+
+                    if usize::try_from((packet.timestamp.0 - user.last_rtp_timestamp).0).ok()?
+                        * std::mem::size_of::<i16>()
+                        == audio.len()
+                    {
+                        // If this recording is the continuation of the previous one, we simply append
+                        if let Some(recording) = user.recordings.back_mut() {
+                            recording.data.extend(audio);
+                        }
+                    } else {
+                        // If it is a new recording, we create a new entry
+                        user.recordings.push_back(VoiceRecording {
+                            timestamp: SystemTime::now(),
+                            data: audio.clone(),
+                        });
                     }
+                    user.last_voice_activity = SystemTime::now();
+                    user.last_rtp_timestamp = packet.timestamp.0;
                 } else {
                     error!("RTP packet, but no audio. Driver may not be configured to decode.");
                 }
@@ -255,6 +215,68 @@ impl GuildRecorderArc {
         }))
     }
 
+    /// Spawns a garbage collector thread for the given user. The thread periodically checks whether the user is
+    /// inactive and removes them if so.
+    #[instrument(skip(self))]
+    fn spawn_user_gc(&self, ssrc: u32) {
+        let users_lock = self.users.clone();
+        let span = span!(Level::INFO, "user_gc", ssrc);
+        tokio::spawn(
+            async move {
+                // Delete the user if no voice activity in the last hour
+                loop {
+                    trace!("Sleeping");
+                    sleep(Duration::from_secs(60 * 60 * 2)).await;
+
+                    let delete;
+                    {
+                        let users = users_lock.read().await;
+                        if let Some(user_mutex) = users.get(&ssrc) {
+                            let user = user_mutex.lock().await;
+                            delete = SystemTime::now()
+                                .duration_since(user.last_voice_activity)
+                                .unwrap()
+                                > Duration::from_secs(60 * 60);
+                        } else {
+                            debug!("User already deleted");
+                            break;
+                        }
+                    }
+
+                    debug!(delete, "Checking timeout of user data");
+
+                    if delete {
+                        let mut users = users_lock.write().await;
+                        users.remove(&ssrc);
+                        debug!("{} remaining known users", users.len());
+                        break;
+                    }
+                }
+            }
+            .instrument(span),
+        );
+    }
+
+    /// Cleans up timed out recordings for a user
+    #[instrument(skip(self, user))]
+    fn cleanup_user_recordings(&self, user: &mut MutexGuard<UserData>) -> Option<()> {
+        while let Some(true) = user.recordings.front().map(|front| {
+            SystemTime::now().duration_since(front.timestamp).unwrap()
+                > Duration::from_secs(60) + samples_to_duration(front.data.len())
+        }) {
+            user.recordings
+                .pop_front()
+                .expect("Missing element in Deque");
+        }
+
+        debug!(
+            remaining_recordings = user.recordings.len(),
+            "Removed timed out recordings"
+        );
+
+        Some(())
+    }
+
     /// Saves the recording to disk
     #[instrument(skip(self, cache_and_http), err)]
     pub async fn save_recording(&self, cache_and_http: &CacheHttp) -> Result<(), RecordingError> {
@@ -263,7 +285,10 @@ impl GuildRecorderArc {
             let users = self.users.read().await;
 
             for user in users.values() {
-                let user = user.lock().await;
+                let mut user = user.lock().await;
+
+                // Make sure we only consider recordings that are within scope
+                self.cleanup_user_recordings(&mut user);
 
                 if !user.recordings.is_empty() {
                     recordings.insert(user.user_id, user.recordings.clone());
@@ -375,7 +400,7 @@ impl GuildRecorderArc {
         ];
         let mut child = Command::new("ffmpeg")
             .kill_on_drop(true)
-            .args(&args)
+            .args(args)
             .arg(file.as_os_str())
             .stdin(Stdio::piped())
             .stdout(Stdio::null())

frontend/angular.json

@@ -119,6 +119,9 @@
     }
   },
   "cli": {
-    "schematicCollections": ["@angular-eslint/schematics"]
+    "schematicCollections": [
+      "@angular-eslint/schematics"
+    ],
+    "analytics": false
   }
 }