From 1733dfdbf91b40420140176bfe631f7d23037765 Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Tue, 8 Sep 2026 19:10:46 -0400 Subject: [PATCH] Stop WhoToFollow from recommending muted or blocked accounts SGS mute/block list hydrators now mark the query ready only on a successful read. For You and Latest Following blender pipelines load those lists. WhoToFollow stays disabled until both lists are ready and sends muted/blocked ids in excluded_user_ids ahead of fatigue ids. --- .../following_candidate_pipeline.rs | 8 ++ .../for_you_candidate_pipeline.rs | 30 +++- home-mixer/models/query.rs | 10 ++ .../blocked_user_ids_query_hydrator.rs | 34 +++++ .../muted_user_ids_query_hydrator.rs | 34 +++++ home-mixer/sources/who_to_follow_source.rs | 134 +++++++++++++++++- 6 files changed, 243 insertions(+), 7 deletions(-) diff --git a/home-mixer/candidate_pipeline/following_candidate_pipeline.rs b/home-mixer/candidate_pipeline/following_candidate_pipeline.rs index c1efd927..4bbcea04 100644 --- a/home-mixer/candidate_pipeline/following_candidate_pipeline.rs +++ b/home-mixer/candidate_pipeline/following_candidate_pipeline.rs @@ -14,7 +14,9 @@ use crate::clients::who_to_follow_client::{ use crate::filters::invalid_conversation_module_filter::InvalidConversationModuleFilter; use crate::models::query::ScoredPostsQuery; use crate::params; +use crate::query_hydrators::blocked_user_ids_query_hydrator::BlockedUserIdsQueryHydrator; use crate::query_hydrators::followed_user_ids_query_hydrator::FollowedUserIdsQueryHydrator; +use crate::query_hydrators::muted_user_ids_query_hydrator::MutedUserIdsQueryHydrator; use crate::query_hydrators::past_request_timestamps_query_hydrator::PastRequestTimestampsQueryHydrator; use crate::query_hydrators::served_history_query_hydrator::ServedHistoryQueryHydrator; use crate::selectors::FollowingBlenderSelector; @@ -168,6 +170,12 @@ impl FollowingCandidatePipeline { Box::new(PastRequestTimestampsQueryHydrator::new(Arc::clone( &past_request_timestamps_client, ))), + Box::new(BlockedUserIdsQueryHydrator { + socialgraph_client: socialgraph_client.clone(), + }), + Box::new(MutedUserIdsQueryHydrator { + socialgraph_client: socialgraph_client.clone(), + }), Box::new(FollowedUserIdsQueryHydrator { socialgraph_client }), ]; diff --git a/home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs b/home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs index 9ad2512d..5fb71e8c 100644 --- a/home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs +++ b/home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs @@ -3,6 +3,7 @@ use crate::clients::past_request_timestamps_client::{ MockPastRequestTimestampsClient, PastRequestTimestampsClient, ProdPastRequestTimestampsClient, }; use crate::clients::prompts_client::{MockPromptsClient, ProdPromptsClient, PromptsClient}; +use crate::clients::s2s::{S2S_CHAIN_PATH, S2S_CRT_PATH, S2S_KEY_PATH}; use crate::clients::served_history_client::{ MockServedHistoryClient, ProdServedHistoryClient, ServedHistoryClient, }; @@ -14,6 +15,8 @@ use crate::filters::ad_adjacent_served_filter::AdAdjacentServedFilter; use crate::filters::push_to_home_dedup_filter::PushToHomeDedupFilter; use crate::models::query::ScoredPostsQuery; use crate::params; +use crate::query_hydrators::blocked_user_ids_query_hydrator::BlockedUserIdsQueryHydrator; +use crate::query_hydrators::muted_user_ids_query_hydrator::MutedUserIdsQueryHydrator; use crate::query_hydrators::past_request_timestamps_query_hydrator::PastRequestTimestampsQueryHydrator; use crate::query_hydrators::served_history_query_hydrator::ServedHistoryQueryHydrator; use crate::scored_posts_server::ScoredPostsServer; @@ -41,8 +44,8 @@ use xai_candidate_pipeline::component_library::clients::kafka_publisher_client:: KafkaPublisherClient, MockKafkaPublisherClient, }; use xai_candidate_pipeline::component_library::clients::{ - MockReplyMixerClient, MockStratoClient, ProdReplyMixerClient, ProdStratoClient, - ReplyMixerClient, StratoClient, + MockReplyMixerClient, MockSocialGraphClient, MockStratoClient, ProdReplyMixerClient, + ProdStratoClient, ReplyMixerClient, SocialGraphClient, SocialGraphClientOps, StratoClient, }; use xai_candidate_pipeline::filter::Filter; use xai_candidate_pipeline::hydrator::Hydrator; @@ -83,6 +86,7 @@ impl ForYouCandidatePipeline { tes_client, reply_mixer_client, strato_client, + socialgraph_client, ) = tokio::join!( async { Arc::new( @@ -145,6 +149,18 @@ impl ForYouCandidatePipeline { .expect("Failed to create Strato client"), ) as Arc }, + async { + Arc::new( + SocialGraphClient::new( + datacenter, + &S2S_CHAIN_PATH, + &S2S_CRT_PATH, + &S2S_KEY_PATH, + ) + .await + .expect("Failed to create flock SocialGraphClient"), + ) as Arc + }, ); Self::build( @@ -162,6 +178,7 @@ impl ForYouCandidatePipeline { past_request_timestamps_client, tes_client, reply_mixer_client, + socialgraph_client, ) } @@ -181,6 +198,7 @@ impl ForYouCandidatePipeline { past_request_timestamps_client: Arc, tes_client: Arc, reply_mixer_client: Arc, + socialgraph_client: Arc, ) -> Self { let query_hydrators: Vec>> = vec![ Box::new(ServedHistoryQueryHydrator::from_client(Arc::clone( @@ -189,6 +207,12 @@ impl ForYouCandidatePipeline { Box::new(PastRequestTimestampsQueryHydrator::new(Arc::clone( &past_request_timestamps_client, ))), + Box::new(BlockedUserIdsQueryHydrator { + socialgraph_client: socialgraph_client.clone(), + }), + Box::new(MutedUserIdsQueryHydrator { + socialgraph_client, + }), ]; let sources: Vec>> = vec![ @@ -263,6 +287,7 @@ impl ForYouCandidatePipeline { let reply_mixer_client: Arc = Arc::new(MockReplyMixerClient); let strato_client: Arc = Arc::new(MockStratoClient::default()); + let socialgraph_client: Arc = Arc::new(MockSocialGraphClient); Self::build( scored_posts_server, strato_client, @@ -278,6 +303,7 @@ impl ForYouCandidatePipeline { past_request_timestamps_client, tes_client, reply_mixer_client, + socialgraph_client, ) } } diff --git a/home-mixer/models/query.rs b/home-mixer/models/query.rs index de645475..4922c1ea 100644 --- a/home-mixer/models/query.rs +++ b/home-mixer/models/query.rs @@ -113,6 +113,10 @@ pub struct ScoredPostsQuery { #[serde(skip)] pub served_history: Vec, pub who_to_follow_eligible: bool, + /// True only after a successful SGS mute-list read. Err leaves this false. + pub muted_user_ids_hydrated: bool, + /// True only after a successful SGS block-list read. Err leaves this false. + pub blocked_user_ids_hydrated: bool, pub feed_survey_eligible: bool, #[serde(serialize_with = "serialize_debug")] pub non_polling_timestamps: Option, @@ -222,6 +226,8 @@ impl ScoredPostsQuery { request_context: String::new(), served_history: vec![], who_to_follow_eligible: false, + muted_user_ids_hydrated: false, + blocked_user_ids_hydrated: false, feed_survey_eligible: false, non_polling_timestamps: None, impressed_post_ids: Vec::new(), @@ -242,6 +248,10 @@ impl ScoredPostsQuery { pub fn has_excluded_topics(&self) -> bool { !self.excluded_topic_ids.is_empty() } + + pub fn mute_block_lists_ready(&self) -> bool { + self.muted_user_ids_hydrated && self.blocked_user_ids_hydrated + } } impl GetTwitterContextViewer for ScoredPostsQuery { diff --git a/home-mixer/query_hydrators/blocked_user_ids_query_hydrator.rs b/home-mixer/query_hydrators/blocked_user_ids_query_hydrator.rs index 870a4599..de19e7c8 100644 --- a/home-mixer/query_hydrators/blocked_user_ids_query_hydrator.rs +++ b/home-mixer/query_hydrators/blocked_user_ids_query_hydrator.rs @@ -11,6 +11,10 @@ pub struct BlockedUserIdsQueryHydrator { #[async_trait] impl QueryHydrator for BlockedUserIdsQueryHydrator { + fn enable(&self, query: &ScoredPostsQuery) -> bool { + !query.blocked_user_ids_hydrated + } + async fn hydrate(&self, query: &ScoredPostsQuery) -> Result { let blocked_user_ids = self .socialgraph_client @@ -23,11 +27,41 @@ impl QueryHydrator for BlockedUserIdsQueryHydrator { blocked_user_ids, ..Default::default() }, + blocked_user_ids_hydrated: true, ..Default::default() }) } fn update(&self, query: &mut ScoredPostsQuery, hydrated: ScoredPostsQuery) { query.user_features.blocked_user_ids = hydrated.user_features.blocked_user_ids; + query.blocked_user_ids_hydrated = true; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use xai_candidate_pipeline::component_library::clients::MockSocialGraphClient; + + #[test] + fn update_marks_list_ready() { + let hydrator = BlockedUserIdsQueryHydrator { + socialgraph_client: Arc::new(MockSocialGraphClient), + }; + let mut query = ScoredPostsQuery::default(); + let mut hydrated = ScoredPostsQuery::default(); + hydrated.user_features.blocked_user_ids = vec![9]; + hydrator.update(&mut query, hydrated); + assert_eq!(query.user_features.blocked_user_ids, vec![9]); + assert!(query.blocked_user_ids_hydrated); + assert!(!hydrator.enable(&query)); + } + + #[test] + fn enable_when_list_not_loaded() { + let hydrator = BlockedUserIdsQueryHydrator { + socialgraph_client: Arc::new(MockSocialGraphClient), + }; + assert!(hydrator.enable(&ScoredPostsQuery::default())); } } diff --git a/home-mixer/query_hydrators/muted_user_ids_query_hydrator.rs b/home-mixer/query_hydrators/muted_user_ids_query_hydrator.rs index d5a447a7..a2b2c10d 100644 --- a/home-mixer/query_hydrators/muted_user_ids_query_hydrator.rs +++ b/home-mixer/query_hydrators/muted_user_ids_query_hydrator.rs @@ -11,6 +11,10 @@ pub struct MutedUserIdsQueryHydrator { #[async_trait] impl QueryHydrator for MutedUserIdsQueryHydrator { + fn enable(&self, query: &ScoredPostsQuery) -> bool { + !query.muted_user_ids_hydrated + } + async fn hydrate(&self, query: &ScoredPostsQuery) -> Result { let muted_user_ids = self .socialgraph_client @@ -23,11 +27,41 @@ impl QueryHydrator for MutedUserIdsQueryHydrator { muted_user_ids, ..Default::default() }, + muted_user_ids_hydrated: true, ..Default::default() }) } fn update(&self, query: &mut ScoredPostsQuery, hydrated: ScoredPostsQuery) { query.user_features.muted_user_ids = hydrated.user_features.muted_user_ids; + query.muted_user_ids_hydrated = true; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use xai_candidate_pipeline::component_library::clients::MockSocialGraphClient; + + #[test] + fn update_marks_list_ready() { + let hydrator = MutedUserIdsQueryHydrator { + socialgraph_client: Arc::new(MockSocialGraphClient), + }; + let mut query = ScoredPostsQuery::default(); + let mut hydrated = ScoredPostsQuery::default(); + hydrated.user_features.muted_user_ids = vec![7]; + hydrator.update(&mut query, hydrated); + assert_eq!(query.user_features.muted_user_ids, vec![7]); + assert!(query.muted_user_ids_hydrated); + assert!(!hydrator.enable(&query)); + } + + #[test] + fn enable_when_list_not_loaded() { + let hydrator = MutedUserIdsQueryHydrator { + socialgraph_client: Arc::new(MockSocialGraphClient), + }; + assert!(hydrator.enable(&ScoredPostsQuery::default())); } } diff --git a/home-mixer/sources/who_to_follow_source.rs b/home-mixer/sources/who_to_follow_source.rs index b5ddefa9..9f69faae 100644 --- a/home-mixer/sources/who_to_follow_source.rs +++ b/home-mixer/sources/who_to_follow_source.rs @@ -1,6 +1,7 @@ use crate::clients::who_to_follow_client::WhoToFollowClient; use crate::models::query::{RequestType, ScoredPostsQuery}; use crate::params::EnableWhoToFollowModule; +use std::collections::HashSet; use std::sync::Arc; use tonic::async_trait; use xai_account_recommendations_mixer_proto::{ @@ -22,7 +23,9 @@ pub struct WhoToFollowSource { #[async_trait] impl Source for WhoToFollowSource { fn enable(&self, query: &ScoredPostsQuery) -> bool { - query.params.get(EnableWhoToFollowModule) && query.who_to_follow_eligible + query.params.get(EnableWhoToFollowModule) + && query.who_to_follow_eligible + && query.mute_block_lists_ready() } async fn source(&self, query: &ScoredPostsQuery) -> Result, String> { @@ -104,7 +107,26 @@ fn build_wtf_request(query: &ScoredPostsQuery) -> AccountRecommendationsMixerReq } fn get_excluded_user_ids(query: &ScoredPostsQuery) -> Vec { - query + let mut ids = Vec::with_capacity(EXCLUDED_USER_IDS_LIMIT); + let mut seen = HashSet::new(); + + // Mute/block first so fatigue IDs cannot crowd a blocked account out of the cap. + for id in query + .user_features + .blocked_user_ids + .iter() + .chain(query.user_features.muted_user_ids.iter()) + .copied() + { + if seen.insert(id) { + ids.push(id); + if ids.len() >= EXCLUDED_USER_IDS_LIMIT { + return ids; + } + } + } + + let fatigue = query .served_history .iter() .flat_map(|sh| &sh.entries) @@ -115,7 +137,109 @@ fn get_excluded_user_ids(query: &ScoredPostsQuery) -> Vec { .iter() .flatten() .filter_map(|item| item.user_id) - }) - .take(EXCLUDED_USER_IDS_LIMIT) - .collect() + }); + + for id in fatigue { + if seen.insert(id) { + ids.push(id); + if ids.len() >= EXCLUDED_USER_IDS_LIMIT { + return ids; + } + } + } + + ids +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::clients::who_to_follow_client::MockWhoToFollowClient; + use crate::models::user_features::UserFeatures; + use xai_x_thrift::served_history::{ + EntityIdType, EntryWithItemIds, ItemIds, ServedHistory, ServedRequestType, + }; + + fn source() -> WhoToFollowSource { + WhoToFollowSource { + who_to_follow_client: Arc::new(MockWhoToFollowClient), + } + } + + fn fatigue_history(user_id: i64) -> Vec { + vec![ServedHistory { + request_type: ServedRequestType::OLDER, + served_id: None, + served_time_ms: Some(1), + entries: vec![EntryWithItemIds { + entity_type: EntityIdType::WHO_TO_FOLLOW, + sort_index: None, + size: None, + item_ids: Some(vec![ItemIds { + user_id: Some(user_id), + tweet_id: None, + source_tweet_id: None, + quote_tweet_id: None, + source_author_id: None, + quote_author_id: None, + in_reply_to_tweet_id: None, + in_reply_to_author_id: None, + article_id: None, + tweet_score: None, + entry_id_to_replace: None, + impression_id: None, + }]), + }], + }] + } + + #[test] + fn enable_requires_mute_and_block_lists() { + let src = source(); + let mut query = ScoredPostsQuery { + who_to_follow_eligible: true, + ..Default::default() + }; + assert!(!src.enable(&query)); + + query.muted_user_ids_hydrated = true; + assert!(!src.enable(&query)); + + query.blocked_user_ids_hydrated = true; + assert!(src.enable(&query)); + } + + #[test] + fn excludes_muted_and_blocked_before_fatigue() { + let query = ScoredPostsQuery { + user_features: UserFeatures { + blocked_user_ids: vec![10], + muted_user_ids: vec![20], + ..Default::default() + }, + served_history: fatigue_history(30), + ..Default::default() + }; + let ids = get_excluded_user_ids(&query); + assert_eq!(ids, vec![10, 20, 30]); + } + + #[test] + fn mute_block_ids_win_the_cap_over_fatigue() { + let query = ScoredPostsQuery { + user_features: UserFeatures { + blocked_user_ids: (1..=EXCLUDED_USER_IDS_LIMIT as i64).collect(), + muted_user_ids: vec![9_999], + ..Default::default() + }, + served_history: fatigue_history(8_888), + ..Default::default() + }; + let ids = get_excluded_user_ids(&query); + assert_eq!(ids.len(), EXCLUDED_USER_IDS_LIMIT); + assert!(ids.contains(&1)); + assert!(ids.contains(&(EXCLUDED_USER_IDS_LIMIT as i64))); + assert!(!ids.contains(&8_888)); + assert!(!ids.contains(&9_999)); + } }