From 647a3d82173eb120672b78e1a282287fd6f0abaa Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:26:15 -0400 Subject: [PATCH] Fail closed when exclusive TES lookup is Err or missing TES exclusive_controls timeout/error and mixer subscription_author_ids miss were assembled as not-exclusive, so Super Follow posts ranked as normal organic. Stamp exclusive_hydration_failed and drop those ids. --- .../subscription_hydrator.rs | 122 ++++++++++-- .../filters/ineligible_subscription_filter.rs | 35 +++- home-mixer/models/candidate.rs | 2 + visibility-filtering/hydration/batch.rs | 11 ++ .../hydration/exclusive_content_hydrator.rs | 174 +++++++++++++----- visibility-filtering/hydration/mod.rs | 58 +++++- visibility-filtering/models/mod.rs | 3 + visibility-filtering/rules/context.rs | 5 + visibility-filtering/rules/golden_corpus.rs | 13 ++ visibility-filtering/rules/tweet_rules.rs | 25 +++ 10 files changed, 370 insertions(+), 78 deletions(-) diff --git a/home-mixer/candidate_hydrators/subscription_hydrator.rs b/home-mixer/candidate_hydrators/subscription_hydrator.rs index b5309eb3..59c21773 100644 --- a/home-mixer/candidate_hydrators/subscription_hydrator.rs +++ b/home-mixer/candidate_hydrators/subscription_hydrator.rs @@ -6,9 +6,15 @@ use tonic::async_trait; use xai_candidate_pipeline::component_library::utils::{default_quick_cache, QuickCache}; use xai_candidate_pipeline::hydrator::{CacheStore, CachedHydrator}; +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum SubscriptionAuthorHydration { + Resolved(Option), + Failed, +} + pub struct SubscriptionHydrator { pub tes_client: Arc, - pub cache: QuickCache>, + pub cache: QuickCache, } impl SubscriptionHydrator { @@ -21,7 +27,7 @@ impl SubscriptionHydrator { #[async_trait] impl CachedHydrator for SubscriptionHydrator { type CacheKey = u64; - type CacheValue = Option; + type CacheValue = SubscriptionAuthorHydration; fn enable(&self, query: &ScoredPostsQuery) -> bool { !query.has_cached_posts @@ -35,14 +41,11 @@ impl CachedHydrator for SubscriptionHydrator { } fn cache_value(&self, hydrated: &PostCandidate) -> Self::CacheValue { - hydrated.subscription_author_id + cache_value(hydrated) } fn hydrate_from_cache(&self, value: Self::CacheValue) -> PostCandidate { - PostCandidate { - subscription_author_id: value, - ..Default::default() - } + hydrate_from_cache(value) } async fn hydrate_from_client( @@ -58,19 +61,7 @@ impl CachedHydrator for SubscriptionHydrator { let mut hydrated_candidates = Vec::with_capacity(candidates.len()); for tweet_id in tweet_ids { - let post_features = post_features.get(&tweet_id); - let hydrated = match post_features { - Some(Ok(value)) => Ok(PostCandidate { - subscription_author_id: *value, - ..Default::default() - }), - None => Err(format!( - "Missing subscription author id for tweet_id={}", - tweet_id - )), - Some(Err(err)) => Err(err.to_string()), - }; - hydrated_candidates.push(hydrated); + hydrated_candidates.push(Ok(hydrate_from_tes(post_features.get(&tweet_id)))); } hydrated_candidates @@ -78,5 +69,96 @@ impl CachedHydrator for SubscriptionHydrator { fn update(&self, candidate: &mut PostCandidate, hydrated: PostCandidate) { candidate.subscription_author_id = hydrated.subscription_author_id; + candidate.subscription_lookup_failed = hydrated.subscription_lookup_failed; + } +} + +fn cache_value(hydrated: &PostCandidate) -> SubscriptionAuthorHydration { + if hydrated.subscription_lookup_failed == Some(true) { + SubscriptionAuthorHydration::Failed + } else { + SubscriptionAuthorHydration::Resolved(hydrated.subscription_author_id) + } +} + +fn hydrate_from_tes(result: Option<&Result, E>>) -> PostCandidate { + match result { + Some(Ok(value)) => PostCandidate { + subscription_author_id: *value, + subscription_lookup_failed: Some(false), + ..Default::default() + }, + None | Some(Err(_)) => PostCandidate { + subscription_lookup_failed: Some(true), + ..Default::default() + }, + } +} + +fn hydrate_from_cache(value: SubscriptionAuthorHydration) -> PostCandidate { + match value { + SubscriptionAuthorHydration::Failed => PostCandidate { + subscription_lookup_failed: Some(true), + ..Default::default() + }, + SubscriptionAuthorHydration::Resolved(author_id) => PostCandidate { + subscription_author_id: author_id, + subscription_lookup_failed: Some(false), + ..Default::default() + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cache_round_trip_preserves_lookup_failure() { + let failed = PostCandidate { + subscription_lookup_failed: Some(true), + ..Default::default() + }; + let cached = cache_value(&failed); + assert_eq!(cached, SubscriptionAuthorHydration::Failed); + let restored = hydrate_from_cache(cached); + assert_eq!(restored.subscription_lookup_failed, Some(true)); + assert!(restored.subscription_author_id.is_none()); + } + + #[test] + fn tes_err_or_missing_slot_fails_closed() { + let err: Result, &str> = Err("tes unavailable"); + let failed = hydrate_from_tes(Some(&err)); + assert_eq!(failed.subscription_lookup_failed, Some(true)); + assert!(failed.subscription_author_id.is_none()); + + let missing: Option<&Result, &str>> = None; + let missing = hydrate_from_tes(missing); + assert_eq!(missing.subscription_lookup_failed, Some(true)); + + let ok: Result, &str> = Ok(None); + let not_exclusive = hydrate_from_tes(Some(&ok)); + assert_eq!(not_exclusive.subscription_lookup_failed, Some(false)); + assert!(not_exclusive.subscription_author_id.is_none()); + + let exclusive: Result, &str> = Ok(Some(42)); + let exclusive = hydrate_from_tes(Some(&exclusive)); + assert_eq!(exclusive.subscription_lookup_failed, Some(false)); + assert_eq!(exclusive.subscription_author_id, Some(42)); + } + + #[test] + fn cache_round_trip_preserves_not_exclusive() { + let resolved = PostCandidate { + subscription_author_id: None, + subscription_lookup_failed: Some(false), + ..Default::default() + }; + let cached = cache_value(&resolved); + assert_eq!(cached, SubscriptionAuthorHydration::Resolved(None)); + let restored = hydrate_from_cache(cached); + assert_eq!(restored.subscription_lookup_failed, Some(false)); + assert!(restored.subscription_author_id.is_none()); } } diff --git a/home-mixer/filters/ineligible_subscription_filter.rs b/home-mixer/filters/ineligible_subscription_filter.rs index 39fe76b4..f93e32cc 100644 --- a/home-mixer/filters/ineligible_subscription_filter.rs +++ b/home-mixer/filters/ineligible_subscription_filter.rs @@ -19,17 +19,27 @@ impl Filter for IneligibleSubscriptionFilter { .collect(); let (kept, removed): (Vec<_>, Vec<_>) = - candidates - .into_iter() - .partition(|candidate| match candidate.subscription_author_id { - Some(author_id) => subscribed_user_ids.contains(&author_id), - None => true, - }); + candidates.into_iter().partition(|candidate| { + keep_subscription_candidate(candidate, &subscribed_user_ids) + }); FilterResult { kept, removed } } } +fn keep_subscription_candidate( + candidate: &PostCandidate, + subscribed_user_ids: &HashSet, +) -> bool { + if candidate.subscription_lookup_failed == Some(true) { + return false; + } + match candidate.subscription_author_id { + Some(author_id) => subscribed_user_ids.contains(&author_id), + None => true, + } +} + #[cfg(test)] mod tests { use super::*; @@ -66,4 +76,17 @@ mod tests { .iter() .any(|c| c.subscription_author_id == Some(3))); } + + #[test] + fn tes_lookup_failure_drops_even_when_not_stamped_exclusive() { + let failed = PostCandidate { + subscription_author_id: None, + subscription_lookup_failed: Some(true), + ..Default::default() + }; + let subscribed = HashSet::from([1u64, 2]); + assert!(!keep_subscription_candidate(&failed, &subscribed)); + assert!(keep_subscription_candidate(&candidate(None), &subscribed)); + assert!(keep_subscription_candidate(&candidate(Some(1)), &subscribed)); + } } diff --git a/home-mixer/models/candidate.rs b/home-mixer/models/candidate.rs index a9d747d6..8c29de40 100644 --- a/home-mixer/models/candidate.rs +++ b/home-mixer/models/candidate.rs @@ -47,6 +47,8 @@ pub struct PostCandidate { pub visibility_reason: Option, pub drop_ancillary_posts: Option, pub subscription_author_id: Option, + #[serde(default)] + pub subscription_lookup_failed: Option, pub tweet_type_metrics: Option>, pub author_blocks_viewer: Option, pub quoted_author_blocks_viewer: Option, diff --git a/visibility-filtering/hydration/batch.rs b/visibility-filtering/hydration/batch.rs index 4f9dc9ac..03f41444 100644 --- a/visibility-filtering/hydration/batch.rs +++ b/visibility-filtering/hydration/batch.rs @@ -102,6 +102,13 @@ impl HydrationBatch { } } + pub(crate) fn is_failed(&self, key: &K) -> bool { + match self.results.get(key) { + Some(hydrated) => hydrated.is_failed(), + None => true, + } + } + pub(crate) fn hydrated(&self, key: &K) -> Option<&Hydrated> { self.results.get(key) } @@ -202,6 +209,10 @@ mod tests { ); assert_eq!(batch.get_or_default(&2), 0); assert_eq!(batch.get_or_default(&3), 0); + assert!(!batch.is_failed(&1)); + assert!(!batch.is_failed(&2)); + assert!(batch.is_failed(&3)); + assert!(batch.is_failed(&99)); } #[test] diff --git a/visibility-filtering/hydration/exclusive_content_hydrator.rs b/visibility-filtering/hydration/exclusive_content_hydrator.rs index d5a2b745..97b135fa 100644 --- a/visibility-filtering/hydration/exclusive_content_hydrator.rs +++ b/visibility-filtering/hydration/exclusive_content_hydrator.rs @@ -1,5 +1,6 @@ use crate::clients::socialgraph_client::SocialgraphClient; -use crate::hydration::metrics::{batch_outcome, record_batch_size, timed_rpc, HydratorOutcome}; +use crate::hydration::batch::{Hydrated, TweetHydrationBatch}; +use crate::hydration::metrics::{record_batch_size, timed_results}; use crate::models::{ExclusiveContentFeatures, TweetId, Viewer}; use crate::rules::SafetyLevel; use std::collections::{HashMap, HashSet}; @@ -21,29 +22,33 @@ impl ExclusiveContentHydrator { tweet_ids: &[TweetId], viewer: Viewer, safety_level: SafetyLevel, - ) -> HashMap> { - let raw_ids: Vec = tweet_ids.iter().map(|t| t.0).collect(); - let candidate_count = raw_ids.len(); - record_batch_size(CLIENT, candidate_count); + ) -> TweetHydrationBatch { + let candidate_count_by_key: HashMap = tweet_ids + .iter() + .fold(HashMap::with_capacity(tweet_ids.len()), |mut counts, id| { + *counts.entry(id.0).or_default() += 1; + counts + }); + record_batch_size(CLIENT, candidate_count_by_key.len()); + let raw_ids: Vec = candidate_count_by_key.keys().copied().collect(); - let exclusive_controls = timed_rpc( + let exclusive_controls = timed_results( CLIENT, "get_exclusive_controls", safety_level, - candidate_count, + &candidate_count_by_key, CLIENT_TIMEOUT, - batch_outcome, self.tes_client.get_exclusive_controls(raw_ids), ) - .await; - - let root_author_ids: Vec = exclusive_controls - .values() - .filter_map(|r| { - r.as_ref() - .ok() - .and_then(|opt| opt.as_ref()) - .map(|ctrl| ctrl.conversation_author_id) + .await + .map_keys(TweetId); + + let root_author_ids: Vec = tweet_ids + .iter() + .filter_map(|id| { + exclusive_controls + .get(id) + .map(|ctrl| ctrl.conversation_author_id as u64) }) .collect::>() .into_iter() @@ -51,40 +56,119 @@ impl ExclusiveContentHydrator { let super_follows = match viewer.user_id() { Some(vid) if !root_author_ids.is_empty() => { - timed_rpc( - CLIENT, - "batch_check_super_follows", + timed_rpc_super_follows( + self.sg_client.as_ref(), + vid, + &root_author_ids, safety_level, - candidate_count, - CLIENT_TIMEOUT, - |_| HydratorOutcome::Success, - self.sg_client - .batch_check_super_follows(vid, &root_author_ids), ) .await } _ => HashMap::new(), }; - tweet_ids - .iter() - .map(|tweet_id| { - let features = exclusive_controls - .get(&tweet_id.0) - .and_then(|r| r.as_ref().ok()) - .and_then(|opt| opt.as_ref()) - .map(|ctrl| { - let author_id = ctrl.conversation_author_id; - ExclusiveContentFeatures { - conversation_author_id: author_id, - viewer_super_follows_author: super_follows - .get(&author_id) - .copied() - .unwrap_or(false), - } - }); - (*tweet_id, features) - }) - .collect() + exclusive_controls.map(|ctrl| { + let author_id = ctrl.conversation_author_id as u64; + ExclusiveContentFeatures { + conversation_author_id: author_id, + viewer_super_follows_author: super_follows.get(&author_id).copied().unwrap_or(false), + } + }) + } +} + +async fn timed_rpc_super_follows( + sg_client: &(dyn SocialgraphClient + Send + Sync), + viewer_id: u64, + root_author_ids: &[u64], + safety_level: SafetyLevel, +) -> HashMap { + use crate::hydration::metrics::{timed_rpc, HydratorOutcome}; + timed_rpc( + CLIENT, + "batch_check_super_follows", + safety_level, + root_author_ids.len(), + CLIENT_TIMEOUT, + |_| HydratorOutcome::Success, + sg_client.batch_check_super_follows(viewer_id, root_author_ids), + ) + .await +} + +pub(crate) fn exclusive_slot( + batch: &TweetHydrationBatch, + tweet_id: &TweetId, +) -> (Option, bool) { + match batch.hydrated(tweet_id) { + Some(Hydrated::Found(features)) => (Some(features.clone()), false), + Some(Hydrated::NotFound) => (None, false), + Some(Hydrated::Failed(_)) | None => (None, true), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hydration::batch::HydrationBatch; + + fn batch( + results: HashMap, &'static str>>, + ) -> TweetHydrationBatch { + let keys: Vec = results.keys().copied().collect(); + HydrationBatch::from_results(keys, results) + } + + fn exclusive(author: u64) -> ExclusiveContentFeatures { + ExclusiveContentFeatures { + conversation_author_id: author, + viewer_super_follows_author: false, + } + } + + #[test] + fn found_exclusive_is_not_a_hydration_failure() { + let batch = batch(HashMap::from([(TweetId(1), Ok(Some(exclusive(100))))])); + let (features, failed) = exclusive_slot(&batch, &TweetId(1)); + assert_eq!(features.unwrap().conversation_author_id, 100); + assert!(!failed); + } + + #[test] + fn not_found_means_not_exclusive() { + let batch = batch(HashMap::from([( + TweetId(1), + Ok::<_, &str>(None), + )])); + let (features, failed) = exclusive_slot(&batch, &TweetId(1)); + assert!(features.is_none()); + assert!(!failed); + } + + #[test] + fn rpc_error_fails_closed() { + let batch = batch(HashMap::from([( + TweetId(1), + Err("tes exclusive unavailable"), + )])); + let (features, failed) = exclusive_slot(&batch, &TweetId(1)); + assert!(features.is_none()); + assert!(failed); + } + + #[test] + fn missing_key_fails_closed() { + let batch = batch(HashMap::from([(TweetId(1), Ok(Some(exclusive(100))))])); + let (features, failed) = exclusive_slot(&batch, &TweetId(2)); + assert!(features.is_none()); + assert!(failed); + } + + #[test] + fn timeout_batch_fails_closed() { + let batch: TweetHydrationBatch = + HydrationBatch::timed_out([TweetId(1), TweetId(2)]); + assert!(exclusive_slot(&batch, &TweetId(1)).1); + assert!(exclusive_slot(&batch, &TweetId(2)).1); } } diff --git a/visibility-filtering/hydration/mod.rs b/visibility-filtering/hydration/mod.rs index 0ea33a16..bd51f099 100644 --- a/visibility-filtering/hydration/mod.rs +++ b/visibility-filtering/hydration/mod.rs @@ -18,7 +18,7 @@ use crate::models::{ use crate::rules::SafetyLevel; use crate::safety_label_source::SafetyLabelSource; use batch::TweetHydrationBatch; -use exclusive_content_hydrator::ExclusiveContentHydrator; +use exclusive_content_hydrator::{exclusive_slot, ExclusiveContentHydrator}; use fallback_cache::FallbackCache; use gizmoduck_hydrator::GizmoduckAuthorHydrator; use safety_label_hydrator::{SafetyLabelHydration, SafetyLabelHydrator}; @@ -59,7 +59,7 @@ struct CandidateFeatures { author_features: TweetHydrationBatch, safety_labels: HashMap, relationships: TweetHydrationBatch, - exclusive_content: HashMap>, + exclusive_content: crate::hydration::batch::TweetHydrationBatch, } impl CandidateFeatures { @@ -67,6 +67,8 @@ impl CandidateFeatures { candidates .iter() .map(|c| { + let (exclusive_content, exclusive_hydration_failed) = + exclusive_slot(&self.exclusive_content, &c.tweet_id); assemble( c, self.tweet_features @@ -79,10 +81,8 @@ impl CandidateFeatures { .cloned() .unwrap_or_default(), self.relationships.get_or_default(&c.tweet_id), - self.exclusive_content - .get(&c.tweet_id) - .cloned() - .unwrap_or_default(), + exclusive_content, + exclusive_hydration_failed, ) }) .collect() @@ -295,7 +295,13 @@ mod tests { })), )]), ), - exclusive_content: HashMap::new(), + exclusive_content: TweetHydrationBatch::from_results( + [TweetId(2)], + HashMap::from([( + TweetId(2), + Ok::, anyhow::Error>(None), + )]), + ), }; let assembled = results.assemble(&candidates); @@ -307,6 +313,44 @@ mod tests { assert_eq!(c.tweet_features.core.source_tweet_id, Some(2)); assert!(c.author_features.is_suspended); assert!(c.relationship.viewer_follows_author); + assert!(!c.exclusive_hydration_failed); + assert!(c.exclusive_content.is_none()); + } + + #[test] + fn assemble_marks_failed_exclusive_lookup() { + let resolved = resolve_candidate(&raw(1, Some(100)), &HashMap::new()).unwrap(); + let results = CandidateFeatures { + tweet_features: HashMap::new(), + author_features: TweetHydrationBatch::from_results( + [TweetId(1)], + HashMap::from([( + TweetId(1), + Ok::<_, anyhow::Error>(Some(AuthorFeatures::default())), + )]), + ), + safety_labels: HashMap::from([(TweetId(1), SafetyLabelMap::default())]), + relationships: TweetHydrationBatch::from_results( + [TweetId(1)], + HashMap::from([( + TweetId(1), + Ok::<_, anyhow::Error>(Some(ViewerAuthorRelationship::default())), + )]), + ), + exclusive_content: TweetHydrationBatch::from_results( + [TweetId(1)], + HashMap::from([( + TweetId(1), + Err::, _>("tes exclusive timeout"), + )]), + ), + }; + + let assembled = results.assemble(&[resolved]); + + assert_eq!(assembled.len(), 1); + assert!(assembled[0].exclusive_hydration_failed); + assert!(assembled[0].exclusive_content.is_none()); } fn raw(tweet_id: u64, request_author_id: Option) -> RawCandidate { diff --git a/visibility-filtering/models/mod.rs b/visibility-filtering/models/mod.rs index b00cd656..78bdc947 100644 --- a/visibility-filtering/models/mod.rs +++ b/visibility-filtering/models/mod.rs @@ -72,6 +72,7 @@ pub struct HydratedTweetCandidate { pub safety_labels: SafetyLabelMap, pub relationship: ViewerAuthorRelationship, pub exclusive_content: Option, + pub exclusive_hydration_failed: bool, } impl HydratedTweetCandidate { @@ -144,6 +145,7 @@ pub fn assemble( safety_labels: SafetyLabelMap, relationship: ViewerAuthorRelationship, exclusive_content: Option, + exclusive_hydration_failed: bool, ) -> HydratedTweetCandidate { HydratedTweetCandidate { tweet_id: candidate.tweet_id.0, @@ -153,6 +155,7 @@ pub fn assemble( safety_labels, relationship, exclusive_content, + exclusive_hydration_failed, } } diff --git a/visibility-filtering/rules/context.rs b/visibility-filtering/rules/context.rs index 5b9b939a..d0879621 100644 --- a/visibility-filtering/rules/context.rs +++ b/visibility-filtering/rules/context.rs @@ -192,6 +192,11 @@ impl TweetPredicates<'_> { pub fn is_exclusive(&self) -> bool { self.ctx.candidate.exclusive_content.is_some() } + + #[inline] + pub fn exclusive_hydration_failed(&self) -> bool { + self.ctx.candidate.exclusive_hydration_failed + } } #[derive(Clone, Copy)] diff --git a/visibility-filtering/rules/golden_corpus.rs b/visibility-filtering/rules/golden_corpus.rs index cc4da723..53982640 100644 --- a/visibility-filtering/rules/golden_corpus.rs +++ b/visibility-filtering/rules/golden_corpus.rs @@ -695,6 +695,19 @@ fn exclusive_content_cases() -> Vec { expected_action: Drop(FilteredReason::ExclusiveTweet), expected_decided_by: Some("DropExclusiveTweetContentRule"), }, + Case { + name: "exclusive_tes_lookup_failed_drops", + level: TimelineHome, + viewer: viewer(VIEWER_ID), + candidate: { + let mut c = exclusive_candidate(false); + c.exclusive_content = None; + c.exclusive_hydration_failed = true; + c + }, + expected_action: Drop(FilteredReason::ExclusiveTweet), + expected_decided_by: Some("DropExclusiveTweetContentRule"), + }, ] } diff --git a/visibility-filtering/rules/tweet_rules.rs b/visibility-filtering/rules/tweet_rules.rs index 15db052e..e04c119e 100644 --- a/visibility-filtering/rules/tweet_rules.rs +++ b/visibility-filtering/rules/tweet_rules.rs @@ -155,6 +155,10 @@ pub(super) const OON_TWEET_LABEL_DROPS: &[RuleSpec] = &[ ]; fn drop_exclusive_tweet_content(context: &RuleContext<'_>) -> VfAction { + if context.tweet().exclusive_hydration_failed() { + return VfAction::Drop(FilteredReason::ExclusiveTweet); + } + if !context.tweet().is_exclusive() { return VfAction::Allow; } @@ -563,6 +567,27 @@ mod tests { &retweet, &FilteredReason::ExclusiveTweet, ); + + let mut lookup_failed = candidate().build(); + lookup_failed.exclusive_hydration_failed = true; + assert_drops( + spec, + &viewer(VIEWER_ID), + &lookup_failed, + &FilteredReason::ExclusiveTweet, + ); + assert_drops( + spec, + &author_viewer(), + &lookup_failed, + &FilteredReason::ExclusiveTweet, + ); + assert_drops( + spec, + &logged_out_viewer(), + &lookup_failed, + &FilteredReason::ExclusiveTweet, + ); } fn gating_viewer(age: ViewerAge) -> ViewerFeatures {