Patches to get deadlocking in osu updates fixed

Cargo.lock

@@ -1698,13 +1698,13 @@ dependencies = [
 
 [[package]]
 name = "serenity"
-version = "0.11.6"
+version = "0.11.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d007dc45584ecc47e791f2a9a7cf17bf98ac386728106f111159c846d624be3f"
+checksum = "7a7a89cef23483fc9d4caf2df41e6d3928e18aada84c56abd237439d929622c6"
 dependencies = [
  "async-trait",
  "async-tungstenite",
- "base64 0.13.1",
+ "base64 0.21.4",
  "bitflags 1.3.2",
  "bytes",
  "cfg-if",

youmubot-osu/src/discord/announcer.rs

@@ -29,10 +29,8 @@ pub struct Announcer {
 }
 
 impl Announcer {
-    pub fn new(client: Osu) -> Self {
-        Self {
-            client: Arc::new(client),
-        }
+    pub fn new(client: Arc<Osu>) -> Self {
+        Self { client }
     }
 }
 
@@ -45,10 +43,12 @@ impl youmubot_prelude::Announcer for Announcer {
         channels: MemberToChannels,
     ) -> Result<()> {
         // For each user...
+        let users = {
             let data = d.read().await;
             let data = data.get::<OsuSavedUsers>().unwrap();
+            data.all().await?
+        };
         let now = chrono::Utc::now();
-        let users = data.all().await?;
         users
             .into_iter()
             .map(|mut osu_user| {
@@ -58,7 +58,7 @@ impl youmubot_prelude::Announcer for Announcer {
                     c: c.clone(),
                     data: d.clone(),
                 };
-                let s = &self;
+                let s = &*self;
                 async move {
                     let channels = channels.channels_of(ctx.c.clone(), user_id).await;
                     if channels.is_empty() {
@@ -76,7 +76,16 @@ impl youmubot_prelude::Announcer for Announcer {
                         Ok(v) => {
                             osu_user.last_update = now;
                             osu_user.pp = v.try_into().unwrap();
-                            data.save(osu_user).await.pls_ok();
+                            let id = osu_user.id;
+                            ctx.data
+                                .read()
+                                .await
+                                .get::<OsuSavedUsers>()
+                                .unwrap()
+                                .save(osu_user)
+                                .await
+                                .pls_ok();
+                            println!("updating {} done", id);
                         }
                         Err(e) => {
                             eprintln!("osu: Cannot update {}: {}", osu_user.id, e);

youmubot-osu/src/discord/mod.rs

@@ -71,25 +71,24 @@ pub fn setup(
 
     // API client
     let http_client = data.get::<HTTPClient>().unwrap().clone();
-    let make_client = || {
-        OsuHttpClient::new(
+    let osu_client = Arc::new(OsuHttpClient::new(
         std::env::var("OSU_API_KEY").expect("Please set OSU_API_KEY as osu! api key."),
-        )
-    };
-    let osu_client = Arc::new(make_client());
+        http_client.clone(),
+    ));
     data.insert::<OsuClient>(osu_client.clone());
     data.insert::<oppai_cache::BeatmapCache>(oppai_cache::BeatmapCache::new(
         http_client,
         sql_client.clone(),
     ));
     data.insert::<beatmap_cache::BeatmapMetaCache>(beatmap_cache::BeatmapMetaCache::new(
-        osu_client, sql_client,
+        osu_client.clone(),
+        sql_client,
     ));
 
     // Announcer
     announcers.add(
         announcer::ANNOUNCER_KEY,
-        announcer::Announcer::new(make_client()),
+        announcer::Announcer::new(osu_client),
     );
     Ok(())
 }

youmubot-osu/src/lib.rs

@@ -33,9 +33,9 @@ fn vec_try_into<U, T: std::convert::TryFrom<U>>(v: Vec<U>) -> Result<Vec<T>, T::
 
 impl Client {
     /// Create a new client from the given API key.
-    pub fn new(key: String) -> Client {
+    pub fn new(key: String, client: HTTPClient) -> Client {
         let client = Ratelimit::new(
-            HTTPClient::new(),
+            client,
             REQUESTS_PER_MINUTE,
             std::time::Duration::from_secs(60),
         );

youmubot-prelude/src/announcer.rs

@@ -19,6 +19,7 @@ use serenity::{
     CacheAndHttp,
 };
 use std::{collections::HashMap, sync::Arc};
+use tokio::time::{interval, MissedTickBehavior};
 use youmubot_db::DB;
 
 /// A list of assigned channels for an announcer.
@@ -149,28 +150,32 @@ impl AnnouncerHandler {
 
     /// Start the AnnouncerHandler, looping forever.
     ///
-    /// It will run all the announcers in sequence every *cooldown* seconds.
+    /// It will run all the announcers every *cooldown* seconds.
     pub async fn scan(self, cooldown: std::time::Duration) {
         // First we store all the keys inside the database.
         let keys = self.announcers.keys().cloned().collect::<Vec<_>>();
         self.data.write().await.insert::<Self>(keys.clone());
-        loop {
-            eprintln!("{}: announcer started scanning", chrono::Utc::now());
-            let after = tokio::time::sleep_until(tokio::time::Instant::now() + cooldown);
         join_all(self.announcers.iter().map(|(key, announcer)| {
+            let data = self.data.clone();
+            let cache = self.cache_http.clone();
+            let mut looper = interval(cooldown);
+            looper.set_missed_tick_behavior(MissedTickBehavior::Delay);
+            async move {
+                loop {
                     eprintln!(" - scanning key `{}`", key);
-                Self::announce(self.data.clone(), self.cache_http.clone(), key, announcer).map(
-                    move |v| {
-                        if let Err(e) = v {
+                    match Self::announce(data.clone(), cache.clone(), key, announcer).await {
+                        Err(e) => {
                             eprintln!(" - key `{}`: {:?}", *key, e)
                         }
-                    },
-                )
+                        Ok(()) => {
+                            eprintln!(" - key `{}`: complete", *key)
+                        }
+                    };
+                    looper.tick().await;
+                }
+            }
         }))
         .await;
-            eprintln!("{}: announcer finished scanning", chrono::Utc::now());
-            after.await;
-        }
     }
 }
 

youmubot-prelude/src/setup.rs

@@ -1,5 +1,5 @@
 use serenity::prelude::*;
-use std::path::Path;
+use std::{path::Path, time::Duration};
 
 /// Set up the prelude libraries.
 ///
@@ -22,7 +22,13 @@ pub async fn setup_prelude(
         .expect("SQL database set up");
 
     // Set up the HTTP client.
-    data.insert::<crate::HTTPClient>(reqwest::Client::new());
+    data.insert::<crate::HTTPClient>(
+        reqwest::ClientBuilder::new()
+            .connect_timeout(Duration::from_secs(5))
+            .timeout(Duration::from_secs(60))
+            .build()
+            .expect("Build be able to build HTTP client"),
+    );
 
     // Set up the member cache.
     data.insert::<crate::MemberCache>(std::sync::Arc::new(crate::MemberCache::default()));

youmubot/src/main.rs

@@ -148,7 +148,7 @@ async fn main() {
     #[cfg(feature = "codeforces")]
     println!("codeforces enabled.");
 
-    tokio::spawn(announcers.scan(std::time::Duration::from_secs(120)));
+    tokio::spawn(announcers.scan(std::time::Duration::from_secs(300)));
 
     println!("Starting...");
     if let Err(v) = client.start().await {