diff --git a/candidate-pipeline/candidate_pipeline.rs b/candidate-pipeline/candidate_pipeline.rs index dd602cc6..faee2294 100644 --- a/candidate-pipeline/candidate_pipeline.rs +++ b/candidate-pipeline/candidate_pipeline.rs @@ -306,7 +306,7 @@ where let hydrate_futures = enabled.iter().map(|h| h.run(query, &candidates)); let results = join_all(hydrate_futures).await; for (hydrator, result) in enabled.iter().zip(results) { - hydrator.update_all(&mut candidates, result); + hydrator.apply_hydration(query, &mut candidates, result); } stats.finish_with_size(candidates.len()); candidates diff --git a/candidate-pipeline/hydrator.rs b/candidate-pipeline/hydrator.rs index 2a8cf78e..ba82862a 100644 --- a/candidate-pipeline/hydrator.rs +++ b/candidate-pipeline/hydrator.rs @@ -57,6 +57,19 @@ where } } + /// Write-back after `join_all`. Hydrators hydrate from one shared snapshot, + /// then `run_hydrators` applies each write in vec order. Override when the + /// stamp must read a field another hydrator just wrote. + fn apply_hydration( + &self, + query: &Q, + candidates: &mut [C], + hydrated: Vec>, + ) { + let _ = query; + self.update_all(candidates, hydrated); + } + fn name(&self) -> &'static str { util::short_type_name(type_name_of_val(self)) } diff --git a/home-mixer/candidate_hydrators/in_network_candidate_hydrator.rs b/home-mixer/candidate_hydrators/in_network_candidate_hydrator.rs index fba4caad..41881770 100644 --- a/home-mixer/candidate_hydrators/in_network_candidate_hydrator.rs +++ b/home-mixer/candidate_hydrators/in_network_candidate_hydrator.rs @@ -1,11 +1,22 @@ use crate::models::candidate::PostCandidate; use crate::models::query::ScoredPostsQuery; -use std::collections::HashSet; use tonic::async_trait; use xai_candidate_pipeline::hydrator::Hydrator; pub struct InNetworkCandidateHydrator; +fn stamp_in_network(query: &ScoredPostsQuery, author_id: u64) -> bool { + let viewer_id = query.user_id; + author_id == viewer_id + || query + .user_features + .followed_user_ids + .iter() + .copied() + .map(|id| id as u64) + .any(|id| id == author_id) +} + #[async_trait] impl Hydrator for InNetworkCandidateHydrator { fn enable(&self, query: &ScoredPostsQuery) -> bool { @@ -17,22 +28,11 @@ impl Hydrator for InNetworkCandidateHydrator { query: &ScoredPostsQuery, candidates: &[PostCandidate], ) -> Vec> { - let viewer_id = query.user_id; - let followed_ids: HashSet = query - .user_features - .followed_user_ids - .iter() - .copied() - .map(|id| id as u64) - .collect(); - candidates .iter() .map(|candidate| { - let is_self = candidate.author_id == viewer_id; - let is_in_network = is_self || followed_ids.contains(&candidate.author_id); Ok(PostCandidate { - in_network: Some(is_in_network), + in_network: Some(stamp_in_network(query, candidate.author_id)), ..Default::default() }) }) @@ -42,4 +42,137 @@ impl Hydrator for InNetworkCandidateHydrator { fn update(&self, candidate: &mut PostCandidate, hydrated: PostCandidate) { candidate.in_network = hydrated.in_network; } + + fn apply_hydration( + &self, + query: &ScoredPostsQuery, + candidates: &mut [PostCandidate], + _hydrated: Vec>, + ) { + // run_hydrators join_alls hydrate() against the source snapshot, then + // applies writes in vec order. CoreData.update may have just filled + // author_id. Copying hydrate()'s snapshot stamp would keep OON on a + // followee TweetMixer shipped as author_id = 0. + for candidate in candidates.iter_mut() { + candidate.in_network = Some(stamp_in_network(query, candidate.author_id)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::user_features::UserFeatures; + + fn query_following(viewer_id: u64, followed: Vec) -> ScoredPostsQuery { + ScoredPostsQuery { + user_id: viewer_id, + user_features: UserFeatures { + followed_user_ids: followed, + ..Default::default() + }, + ..Default::default() + } + } + + #[tokio::test] + async fn hydrate_stamps_from_snapshot_author() { + let query = query_following(1, vec![42]); + let hydrator = InNetworkCandidateHydrator; + let candidates = vec![ + PostCandidate { + tweet_id: 1, + author_id: 42, + ..Default::default() + }, + PostCandidate { + tweet_id: 2, + author_id: 99, + ..Default::default() + }, + PostCandidate { + tweet_id: 3, + author_id: 1, + ..Default::default() + }, + ]; + let hydrated = hydrator.hydrate(&query, &candidates).await; + assert_eq!(hydrated[0].as_ref().unwrap().in_network, Some(true)); + assert_eq!(hydrated[1].as_ref().unwrap().in_network, Some(false)); + assert_eq!(hydrated[2].as_ref().unwrap().in_network, Some(true)); + } + + #[tokio::test] + async fn join_all_snapshot_stamp_is_oon_when_author_id_is_zero() { + let query = query_following(1, vec![42]); + let hydrator = InNetworkCandidateHydrator; + let snapshot = [PostCandidate { + tweet_id: 1, + author_id: 0, + ..Default::default() + }]; + let hydrated = hydrator.hydrate(&query, &snapshot).await; + let mut candidate = snapshot[0].clone(); + hydrator.update(&mut candidate, hydrated.into_iter().next().unwrap().unwrap()); + candidate.author_id = 42; + assert_eq!( + candidate.in_network, + Some(false), + "hydrate+update copies the pre-TES OON stamp; this is the join_all bug" + ); + } + + #[tokio::test] + async fn apply_hydration_restamps_after_tes_fills_followed_author() { + let query = query_following(1, vec![42]); + let hydrator = InNetworkCandidateHydrator; + let snapshot = [PostCandidate { + tweet_id: 1, + author_id: 0, + ..Default::default() + }]; + let hydrated = hydrator.hydrate(&query, &snapshot).await; + let mut candidates = snapshot.to_vec(); + // CoreData.update in the join_all write-back, before InNetwork. + candidates[0].author_id = 42; + hydrator.apply_hydration(&query, &mut candidates, hydrated); + assert_eq!(candidates[0].in_network, Some(true)); + } + + #[tokio::test] + async fn apply_hydration_keeps_oon_for_unfollowed_tes_author() { + let query = query_following(1, vec![42]); + let hydrator = InNetworkCandidateHydrator; + let snapshot = [PostCandidate { + tweet_id: 1, + author_id: 0, + in_network: Some(true), + ..Default::default() + }]; + let hydrated = hydrator.hydrate(&query, &snapshot).await; + let mut candidates = snapshot.to_vec(); + candidates[0].author_id = 99; + hydrator.apply_hydration(&query, &mut candidates, hydrated); + assert_eq!(candidates[0].in_network, Some(false)); + } + + #[tokio::test] + async fn apply_hydration_marks_self_in_network() { + let query = query_following(7, vec![42]); + let hydrator = InNetworkCandidateHydrator; + let mut candidates = vec![PostCandidate { + tweet_id: 1, + author_id: 7, + in_network: Some(false), + ..Default::default() + }]; + hydrator.apply_hydration(&query, &mut candidates, vec![]); + assert_eq!(candidates[0].in_network, Some(true)); + } + + #[test] + fn tes_miss_leaving_author_zero_stamps_oon() { + let query = query_following(1, vec![42]); + assert!(!stamp_in_network(&query, 0)); + } } diff --git a/home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs b/home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs index e8c9c397..e01b0d42 100644 --- a/home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs +++ b/home-mixer/candidate_pipeline/phoenix_candidate_pipeline.rs @@ -324,12 +324,15 @@ impl PhoenixCandidatePipeline { cached_posts_source, ]; + // InNetwork.apply_hydration reads author_id after CoreData.update. + // join_all hydrates every hydrator from the same snapshot, so vec + // order alone cannot give InNetwork the TES author (#128 leftover). let hydrators: Vec>> = vec![ - Box::new(InNetworkCandidateHydrator), Box::new(BidirectionalFollowHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(core_data_hydrator), + Box::new(InNetworkCandidateHydrator), Box::new(QuoteHydrator::new(tes_client.clone(), socialgraph_client.clone()).await), Box::new(MediaInfoHydrator::new(media_info_cache_client).await), Box::new(SubscriptionHydrator::new(tes_client.clone()).await), diff --git a/home-mixer/candidate_pipeline/phoenix_scores_pipeline.rs b/home-mixer/candidate_pipeline/phoenix_scores_pipeline.rs index 6a79621f..69e5fa82 100644 --- a/home-mixer/candidate_pipeline/phoenix_scores_pipeline.rs +++ b/home-mixer/candidate_pipeline/phoenix_scores_pipeline.rs @@ -135,6 +135,8 @@ impl PhoenixScoresPipeline { let sources: Vec>> = vec![Box::new(SeedCandidatesSource)]; + // CoreData.update before InNetwork.apply_hydration. join_all still + // hydrates both from the source snapshot; the stamp is the write-back. let hydrators: Vec>> = vec![ Box::new(CoreDataCandidateHydrator::new(tes_client.clone()).await), Box::new(InNetworkCandidateHydrator),