Merge pull request #32 from blacksky-algorithms/fix-ci

Fix CI: ignore tests that require an active database connection + cargo fmt on various crates

Author: rudyfraser

rsky-common/src/lib.rs

@@ -1,2 +1,2 @@
 pub mod env;
-pub mod explicit_slurs;
+pub mod explicit_slurs;

rsky-feedgen/src/apis/mod.rs

@@ -1,6 +1,4 @@
-use rsky_common::env::env_int;
 use crate::db::*;
-use rsky_common::explicit_slurs::contains_explicit_slurs;
 use crate::models::create_request::CreateRecord;
 use crate::models::Lexicon::{AppBskyFeedFollow, AppBskyFeedLike, AppBskyFeedPost};
 use crate::models::*;
@@ -15,6 +13,8 @@ use once_cell::sync::Lazy;
 use rand::Rng;
 use regex::Regex;
 use rocket::State;
+use rsky_common::env::env_int;
+use rsky_common::explicit_slurs::contains_explicit_slurs;
 use rsky_lexicon::app::bsky::embed::{Embeds, MediaUnion};
 use rsky_lexicon::app::bsky::feed::PostLabels;
 use std::collections::HashSet;
@@ -1109,6 +1109,7 @@ pub async fn get_cursor(
 mod tests {
     use super::*;
     use crate::routes::{index, BLACKSKY};
+    use crate::{ReadReplicaConn1, ReadReplicaConn2};
     use rocket::figment::map;
     use rocket::figment::value::{Map, Value};
     use rocket::http::Status;
@@ -1120,33 +1121,46 @@ mod tests {
     fn before(config: FeedGenConfig) -> Rocket<Build> {
         // Initialize Rocket with ReadReplicaConn and other necessary fairings/routes
         let write_database_url = env::var("DATABASE_URL").unwrap();
-        let read_database_url = env::var("READ_REPLICA_URL").unwrap();
+        let read_database_url_1 = env::var("READ_REPLICA_URL_1").unwrap();
+        let read_database_url_2 = env::var("READ_REPLICA_URL_2").unwrap();
 
         let write_db: Map<_, Value> = map! {
             "url" => write_database_url.into(),
             "pool_size" => 20.into(),
             "timeout" => 30.into(),
         };
 
-        let read_db: Map<_, Value> = map! {
-            "url" => read_database_url.into(),
+        let read_db_1: Map<_, Value> = map! {
+            "url" => read_database_url_1.into(),
+            "pool_size" => 20.into(),
+            "timeout" => 30.into(),
+        };
+
+        let read_db_2: Map<_, Value> = map! {
+            "url" => read_database_url_2.into(),
             "pool_size" => 20.into(),
             "timeout" => 30.into(),
         };
 
         let figment = rocket::Config::figment().merge((
             "databases",
-            map!["pg_read_replica" => read_db, "pg_db" => write_db],
+            map![
+                "pg_read_replica_1" => read_db_1,
+                "pg_read_replica_2" => read_db_2,
+                "pg_db" => write_db
+            ],
         ));
 
         rocket::custom(figment)
             .attach(WriteDbConn::fairing())
-            .attach(ReadReplicaConn::fairing())
+            .attach(ReadReplicaConn1::fairing())
+            .attach(ReadReplicaConn2::fairing())
             .mount("/", routes![index])
             .manage(config)
     }
 
     #[rocket::async_test]
+    #[ignore]
     async fn test_no_sponsored_post_when_show_sponsored_post_is_false() {
         // Set environment variables temporarily using temp_env for this test
         async_with_vars(
@@ -1165,6 +1179,7 @@ mod tests {
                     show_sponsored_post: false,
                     sponsored_post_uri: "at://did:example/sponsored-post".to_string(),
                     sponsored_post_probability: 1.0,
+                    trending_percentile_min: 0.9,
                 };
                 let rocket = before(config.clone());
 
@@ -1205,6 +1220,7 @@ mod tests {
     }
 
     #[rocket::async_test]
+    #[ignore]
     async fn test_sponsored_post_always_returned_when_probability_is_1() {
         async_with_vars(
             vec![
@@ -1222,6 +1238,7 @@ mod tests {
                     show_sponsored_post: true,
                     sponsored_post_uri: "at://did:example/sponsored-post".to_string(),
                     sponsored_post_probability: 1.0,
+                    trending_percentile_min: 0.9,
                 };
                 let rocket = before(config.clone());
 
@@ -1261,6 +1278,7 @@ mod tests {
     }
 
     #[rocket::async_test]
+    #[ignore]
     async fn test_sponsored_post_never_returned_when_limit_is_2() {
         async_with_vars(
             vec![
@@ -1278,6 +1296,7 @@ mod tests {
                     show_sponsored_post: true,
                     sponsored_post_uri: "at://did:example/sponsored-post".to_string(),
                     sponsored_post_probability: 1.0,
+                    trending_percentile_min: 0.9,
                 };
                 let rocket = before(config.clone());
 
@@ -1318,6 +1337,7 @@ mod tests {
     }
 
     #[rocket::async_test]
+    #[ignore]
     async fn test_sponsored_post_returned_50_percent_of_the_time() {
         async_with_vars(
             vec![
@@ -1335,6 +1355,7 @@ mod tests {
                     show_sponsored_post: true,
                     sponsored_post_uri: "at://did:example/sponsored-post".to_string(),
                     sponsored_post_probability: 0.5,
+                    trending_percentile_min: 0.9,
                 };
                 let rocket = before(config.clone());
 
@@ -1387,6 +1408,7 @@ mod tests {
     }
 
     #[rocket::async_test]
+    #[ignore]
     async fn test_sponsored_post_never_last() {
         async_with_vars(
             vec![
@@ -1404,6 +1426,7 @@ mod tests {
                     show_sponsored_post: true,
                     sponsored_post_uri: "at://did:example/sponsored-post".to_string(),
                     sponsored_post_probability: 1.0,
+                    trending_percentile_min: 0.9,
                 };
                 let rocket = before(config.clone());
 

rsky-labeler/src/main.rs

@@ -124,7 +124,7 @@ async fn get_labels(
 
 async fn label_subject(
     agent: &AtpAgent<MemorySessionStore, ReqwestClient>,
-    subject_ref: EmitEventInputSubjectRefs
+    subject_ref: EmitEventInputSubjectRefs,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let label_result = agent
         .api
@@ -140,7 +140,8 @@ async fn label_subject(
                     ModEventLabel {
                         data: ModEventLabelData {
                             comment: Some(
-                                env::var("MOD_SERVICE_LABEL_REASON").unwrap_or("Explicit slur filter".to_string()),
+                                env::var("MOD_SERVICE_LABEL_REASON")
+                                    .unwrap_or("Explicit slur filter".to_string()),
                             ),
                             create_label_vals: vec![env::var("MOD_SERVICE_LABEL")
                                 .unwrap_or("antiblack-harassment".to_string())],
@@ -193,7 +194,7 @@ async fn tag_subject(
 async fn create_report(
     agent: &AtpAgent<MemorySessionStore, ReqwestClient>,
     subject_ref: EmitEventInputSubjectRefs,
-    reason: Option<String>
+    reason: Option<String>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let report_result = agent
         .api
@@ -362,17 +363,21 @@ async fn process(
                         continue;
                     }
                     if env_bool("ENABLE_CREATE_REPORT").unwrap_or(true) {
-                        match create_report(agent, auto_report.subject_ref.clone(), auto_report.reason).await {
+                        match create_report(
+                            agent,
+                            auto_report.subject_ref.clone(),
+                            auto_report.reason,
+                        )
+                        .await
+                        {
                             Ok(()) => (),
                             Err(error) => {
                                 eprintln!("@LOG: Failed to create report for record: {error:?}")
                             }
                         }
                     }
                     if env_bool("ENABLE_CREATE_LABEL").unwrap_or(true) {
-                        match label_subject(agent, auto_report.subject_ref.clone())
-                            .await
-                        {
+                        match label_subject(agent, auto_report.subject_ref.clone()).await {
                             Ok(()) => (),
                             Err(error) => eprintln!("@LOG: Failed to label record: {error:?}"),
                         }

rsky-lexicon/src/com/atproto/label/mod.rs

@@ -5,7 +5,7 @@ use chrono::{DateTime, Utc};
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SubscribeLabels {
     pub seq: i64,
-    pub labels: Vec<Label>
+    pub labels: Vec<Label>,
 }
 
 /// Metadata tag on an atproto resource (eg, repo or record).

rsky-pds/src/repo/mst/mod.rs

@@ -488,20 +488,26 @@ impl MST {
 
         // Now, unwrapping is safe because we either already had entries
         // or we just set them above
-        Ok(self.entries.as_mut().ok_or_else(|| anyhow::anyhow!("No entries"))?)
+        Ok(self
+            .entries
+            .as_mut()
+            .ok_or_else(|| anyhow::anyhow!("No entries"))?)
     }
 
     pub fn get_entries_ref(&self) -> Result<Vec<NodeEntry>> {
         // If `self.entries` is not populated, hydrate it first
         return match self.entries {
             None => {
                 // Read from storage (block store) to get the node data
-                let data: CborValue = self.storage.clone().read_obj(&self.pointer, |obj: &CborValue| {
-                    match serde_cbor::value::from_value::<NodeData>(obj.clone()) {
-                        Ok(_) => true,
-                        Err(_) => false,
-                    }
-                })?;
+                let data: CborValue =
+                    self.storage
+                        .clone()
+                        .read_obj(&self.pointer, |obj: &CborValue| {
+                            match serde_cbor::value::from_value::<NodeData>(obj.clone()) {
+                                Ok(_) => true,
+                                Err(_) => false,
+                            }
+                        })?;
                 let data: NodeData = serde_cbor::value::from_value(data)?;
 
                 // Compute the layer
@@ -513,11 +519,13 @@ impl MST {
 
                 // Deserialize into self.entries
                 util::deserialize_node_data(&self.storage, &data, layer)
-            },
-            Some(ref entries) => Ok(entries.clone())
-        }
+            }
+            Some(ref entries) => Ok(entries.clone()),
+        };
     }
 
+    // We don't hash the node on every mutation for performance reasons
+    // Instead we keep track of whether the pointer is outdated and only (recursively) calculate when needed
     pub fn get_pointer(&mut self) -> Result<Cid> {
         if !self.outdated_pointer {
             return Ok(self.pointer);
@@ -583,7 +591,7 @@ impl MST {
             bytes: common::struct_to_cbor(data)?,
         })
     }
-    
+
     /// In most cases, we get the layer of a node from a hint on creation
     /// In the case of the topmost node in the tree, we look for a key in the node & determine the layer
     /// In the case where we don't find one, we recurse down until we do.
@@ -1211,9 +1219,7 @@ impl MST {
         for entry in self.walk() {
             match entry {
                 NodeEntry::Leaf(_) => nodes.push(entry),
-                NodeEntry::MST(m) => {
-                    nodes.push(NodeEntry::MST(m))
-                }
+                NodeEntry::MST(m) => nodes.push(NodeEntry::MST(m)),
             }
         }
         Ok(nodes)
@@ -1530,6 +1536,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore]
     fn saves_and_loads_from_blockstore() -> Result<()> {
         let mut storage =
             SqlRepoReader::new(None, "did:example:123456789abcdefghi".to_string(), None);