server: avoid learning the same job multiple times

The current learning implementation was designed with Travis CI in mind,
where a single check was sent when the job ended. Because of that, when
a successful build was detected all the successful job outputs were
learned. This assumption doesn't hold anymore for GitHub Actions though.

GHA sends a check for every job in the matrix, and that causes RLA to
learn a successful job every time one of the following jobs in the same
build finishes, learning it multiple times. For example, if a build
contains 60 jobs, the first job to finish will be learned 60 times.

This fixes the problem by remembering which job IDs we learned before,
and skipping the learning process if the ID was previously seen. Along
with that the code to remember IDs was refactored.

Author: pietroalbini

src/bin/server/worker.rs

@@ -2,39 +2,25 @@ use super::{QueueItem, QueueItemKind};
 
 use crate::rla;
 use crate::rla::ci::{self, BuildCommit, CiPlatform};
-use std::collections::VecDeque;
+use std::collections::{HashSet, VecDeque};
+use std::hash::Hash;
 use std::path::PathBuf;
 use std::str;
 
-// We keep track of the last several unique job IDs. This is because
-// Azure sends us a notification for every individual builder's
-// state (around 70 notifications/job as of this time), but we want
-// to only process a given job once.
-//
-// You might ask -- why is this not a HashSet/HashMap? That would
-// also work, but be a little more complicated to remove things
-// from. We would need to keep track of order somehow to remove the
-// oldest job ID. An attempt at such an API was tried in PR #29, but
-// ultimately scrapped as too complicated.
-//
-// We keep few enough elements in this "set" that a Vec isn't too bad.
-//
-// Note: Don't update this number too high, as we O(n) loop through it on every
-// notification from GitHub (twice).
-const KEEP_IDS: usize = 16;
-
 pub struct Worker {
     debug_post: Option<(String, u32)>,
     index_file: PathBuf,
     index: rla::Index,
     extract_config: rla::extract::Config,
     github: rla::github::Client,
     queue: crossbeam::channel::Receiver<QueueItem>,
-    notified: VecDeque<u64>,
     ci: Box<dyn CiPlatform + Send>,
     repo: String,
     secondary_repos: Vec<String>,
     query_builds_from_primary_repo: bool,
+
+    recently_notified: RecentlySeen<u64>,
+    recently_learned: RecentlySeen<String>,
 }
 
 impl Worker {
@@ -66,12 +52,14 @@ impl Worker {
             index_file,
             extract_config: Default::default(),
             github: rla::github::Client::new()?,
-            notified: VecDeque::new(),
             queue,
             ci,
             repo,
             secondary_repos,
             query_builds_from_primary_repo,
+
+            recently_notified: RecentlySeen::new(32),
+            recently_learned: RecentlySeen::new(256),
         })
     }
 
@@ -133,10 +121,6 @@ impl Worker {
 
         info!("started processing");
 
-        if self.notified.contains(&build_id) {
-            info!("ignoring recently notified build");
-            return Ok(());
-        }
         let query_from = if self.query_builds_from_primary_repo {
             &self.repo
         } else {
@@ -160,14 +144,7 @@ impl Worker {
 
         // Avoid processing the same build multiple times.
         if !outcome.is_passed() {
-            info!("preparing report");
-            self.report_failed(build.as_ref())?;
-
-            info!("marked as notified");
-            self.notified.push_front(build_id);
-            if self.notified.len() > KEEP_IDS {
-                self.notified.pop_back();
-            }
+            self.report_failed(build_id, build.as_ref())?;
         }
         if build.pr_number().is_none() && build.branch_name() == "auto" {
             info!("learning from the log");
@@ -179,8 +156,13 @@ impl Worker {
         Ok(())
     }
 
-    fn report_failed(&mut self, build: &dyn rla::ci::Build) -> rla::Result<()> {
-        debug!("Preparing report...");
+    fn report_failed(&mut self, build_id: u64, build: &dyn rla::ci::Build) -> rla::Result<()> {
+        if self.recently_notified.recently_witnessed(&build_id) {
+            info!("avoided reporting recently notified build");
+            return Ok(());
+        }
+
+        info!("preparing report");
 
         let job = match build.jobs().iter().find(|j| j.outcome().is_failed()) {
             Some(job) => *job,
@@ -301,6 +283,9 @@ impl Worker {
 [I'm a bot](https://github.com/rust-lang/rust-log-analyzer)! I can only do what humans tell me to, so if this was not helpful or you have suggestions for improvements, please ping or otherwise contact **`@rust-lang/infra`**. ([Feature Requests](https://github.com/rust-lang/rust-log-analyzer/issues?q=is%3Aopen+is%3Aissue+label%3Afeature-request))
         "#, opening = opening, html_url = job.html_url(), log_url = pretty_log_url, raw_log_url = raw_log_url, log = extracted))?;
 
+        info!("marked build {} as recently notified", build_id);
+        self.recently_notified.store(build_id);
+
         Ok(())
     }
 
@@ -310,6 +295,11 @@ impl Worker {
                 continue;
             }
 
+            if self.recently_learned.recently_witnessed(&job.id()) {
+                info!("Skipped already processed {}", job);
+                return Ok(());
+            }
+
             debug!("Processing {}...", job);
 
             match ci::download_log(self.ci.as_ref(), *job, self.github.internal()) {
@@ -320,6 +310,7 @@ impl Worker {
                             1,
                         );
                     }
+                    self.recently_learned.store(job.id());
                 }
                 None => {
                     warn!(
@@ -341,3 +332,70 @@ impl Worker {
         Ok(())
     }
 }
+
+/// Keeps track of the recently seen IDs for both the failed build reports and the learned jobs.
+/// Only the most recent IDs are stored, to avoid growing the memory usage endlessly.
+///
+/// Internally this uses both an HashSet to provide fast lookups and a VecDeque to know which old
+/// jobs needs to be removed.
+struct RecentlySeen<T: Clone + Eq + Hash> {
+    size: usize,
+    lookup: HashSet<T>,
+    removal: VecDeque<T>,
+}
+
+impl<T: Clone + Eq + Hash> RecentlySeen<T> {
+    fn new(size: usize) -> Self {
+        Self {
+            size,
+            lookup: HashSet::with_capacity(size),
+            removal: VecDeque::with_capacity(size),
+        }
+    }
+
+    fn recently_witnessed(&self, key: &T) -> bool {
+        self.lookup.contains(key)
+    }
+
+    fn store(&mut self, key: T) {
+        if self.lookup.contains(&key) {
+            return;
+        }
+        if self.removal.len() >= self.size {
+            if let Some(item) = self.removal.pop_back() {
+                self.lookup.remove(&item);
+            }
+        }
+        self.lookup.insert(key.clone());
+        self.removal.push_front(key);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_recently_seen() {
+        let mut recently = RecentlySeen::new(2);
+
+        assert!(!recently.recently_witnessed(&0));
+        assert!(!recently.recently_witnessed(&1));
+        assert!(!recently.recently_witnessed(&2));
+
+        recently.store(0);
+        assert!(recently.recently_witnessed(&0));
+        assert!(!recently.recently_witnessed(&1));
+        assert!(!recently.recently_witnessed(&2));
+
+        recently.store(1);
+        assert!(recently.recently_witnessed(&0));
+        assert!(recently.recently_witnessed(&1));
+        assert!(!recently.recently_witnessed(&2));
+
+        recently.store(2);
+        assert!(!recently.recently_witnessed(&0));
+        assert!(recently.recently_witnessed(&1));
+        assert!(recently.recently_witnessed(&2));
+    }
+}