Fail closed when exclusive-content TES lookup errors

visibility-filtering/hydration/exclusive_content_hydrator.rs

@@ -1,6 +1,8 @@
 use crate::clients::socialgraph_client::SocialgraphClient;
-use crate::hydration::metrics::{batch_outcome, record_batch_size, timed_rpc, HydratorOutcome};
-use crate::models::{ExclusiveContentFeatures, TweetId, Viewer};
+use crate::hydration::metrics::{
+    batch_outcome, record_batch_size, timed_rpc, HydratorOutcome,
+};
+use crate::models::{ExclusiveContentFeatures, ExclusiveHydration, TweetId, Viewer};
 use crate::rules::SafetyLevel;
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
@@ -21,12 +23,12 @@ impl ExclusiveContentHydrator {
         tweet_ids: &[TweetId],
         viewer: Viewer,
         safety_level: SafetyLevel,
-    ) -> HashMap<TweetId, Option<ExclusiveContentFeatures>> {
+    ) -> HashMap<TweetId, ExclusiveHydration> {
         let raw_ids: Vec<u64> = tweet_ids.iter().map(|t| t.0).collect();
         let candidate_count = raw_ids.len();
         record_batch_size(CLIENT, candidate_count);
 
-        let exclusive_controls = timed_rpc(
+        let (outcome, exclusive_controls) = timed_rpc(
             CLIENT,
             "get_exclusive_controls",
             safety_level,
@@ -37,21 +39,28 @@ impl ExclusiveContentHydrator {
         )
         .await;
 
-        let root_author_ids: Vec<u64> = exclusive_controls
-            .values()
-            .filter_map(|r| {
-                r.as_ref()
-                    .ok()
-                    .and_then(|opt| opt.as_ref())
-                    .map(|ctrl| ctrl.conversation_author_id)
+        let authors: HashMap<u64, Result<Option<u64>, String>> = exclusive_controls
+            .into_iter()
+            .map(|(id, result)| {
+                (
+                    id,
+                    result
+                        .map(|opt| opt.map(|ctrl| ctrl.conversation_author_id))
+                        .map_err(|_| "tes exclusive_controls failed".to_string()),
+                )
             })
+            .collect();
+
+        let root_author_ids: Vec<u64> = authors
+            .values()
+            .filter_map(|r| r.as_ref().ok().copied().flatten())
             .collect::<HashSet<_>>()
             .into_iter()
             .collect();
 
         let super_follows = match viewer.user_id() {
             Some(vid) if !root_author_ids.is_empty() => {
-                timed_rpc(
+                let (_, follows) = timed_rpc(
                     CLIENT,
                     "batch_check_super_follows",
                     safety_level,
@@ -61,30 +70,111 @@ impl ExclusiveContentHydrator {
                     self.sg_client
                         .batch_check_super_follows(vid, &root_author_ids),
                 )
-                .await
+                .await;
+                follows
             }
             _ => 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)
+        resolve_exclusive(
+            tweet_ids,
+            outcome == HydratorOutcome::Timeout,
+            &authors,
+            &super_follows,
+        )
+    }
+}
+
+pub(crate) fn resolve_exclusive(
+    tweet_ids: &[TweetId],
+    timed_out: bool,
+    exclusive_authors: &HashMap<u64, Result<Option<u64>, String>>,
+    super_follows: &HashMap<u64, bool>,
+) -> HashMap<TweetId, ExclusiveHydration> {
+    tweet_ids
+        .iter()
+        .map(|tweet_id| {
+            if timed_out {
+                return (*tweet_id, ExclusiveHydration::Failed);
+            }
+            let hydration = match exclusive_authors.get(&tweet_id.0) {
+                Some(Err(_)) => ExclusiveHydration::Failed,
+                Some(Ok(None)) | None => ExclusiveHydration::Public,
+                Some(Ok(Some(author_id))) => {
+                    ExclusiveHydration::Exclusive(ExclusiveContentFeatures {
+                        conversation_author_id: *author_id,
+                        viewer_super_follows_author: super_follows
+                            .get(author_id)
+                            .copied()
+                            .unwrap_or(false),
+                    })
+                }
+            };
+            (*tweet_id, hydration)
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn id(n: u64) -> TweetId {
+        TweetId(n)
+    }
+
+    #[test]
+    fn tes_ok_none_is_public() {
+        let authors = HashMap::from([(1u64, Ok(None))]);
+        let out = resolve_exclusive(&[id(1)], false, &authors, &HashMap::new());
+        assert_eq!(out[&id(1)], ExclusiveHydration::Public);
+    }
+
+    #[test]
+    fn tes_ok_some_is_exclusive() {
+        let authors = HashMap::from([(1u64, Ok(Some(42u64)))]);
+        let follows = HashMap::from([(42u64, false)]);
+        let out = resolve_exclusive(&[id(1)], false, &authors, &follows);
+        assert_eq!(
+            out[&id(1)],
+            ExclusiveHydration::Exclusive(ExclusiveContentFeatures {
+                conversation_author_id: 42,
+                viewer_super_follows_author: false,
             })
-            .collect()
+        );
+    }
+
+    #[test]
+    fn tes_err_is_failed() {
+        let authors = HashMap::from([(1u64, Err("tes unavailable".to_string()))]);
+        let out = resolve_exclusive(&[id(1)], false, &authors, &HashMap::new());
+        assert_eq!(out[&id(1)], ExclusiveHydration::Failed);
+    }
+
+    #[test]
+    fn tes_timeout_fails_the_whole_batch() {
+        let authors = HashMap::new();
+        let out = resolve_exclusive(&[id(1), id(2)], true, &authors, &HashMap::new());
+        assert_eq!(out[&id(1)], ExclusiveHydration::Failed);
+        assert_eq!(out[&id(2)], ExclusiveHydration::Failed);
+    }
+
+    #[test]
+    fn successful_empty_map_stays_public() {
+        let out = resolve_exclusive(&[id(1)], false, &HashMap::new(), &HashMap::new());
+        assert_eq!(out[&id(1)], ExclusiveHydration::Public);
+    }
+
+    #[test]
+    fn subscriber_flag_comes_from_super_follows_map() {
+        let authors = HashMap::from([(1u64, Ok(Some(42u64)))]);
+        let follows = HashMap::from([(42u64, true)]);
+        let out = resolve_exclusive(&[id(1)], false, &authors, &follows);
+        match &out[&id(1)] {
+            ExclusiveHydration::Exclusive(features) => {
+                assert!(features.viewer_super_follows_author);
+            }
+            other => panic!("expected exclusive, got {other:?}"),
+        }
     }
 }

visibility-filtering/hydration/metrics.rs

@@ -236,7 +236,7 @@ pub(crate) async fn timed_rpc<T: Default>(
     timeout: Duration,
     classify: impl FnOnce(&T) -> HydratorOutcome,
     fut: impl Future<Output = T>,
-) -> T {
+) -> (HydratorOutcome, T) {
     let start = Instant::now();
     let (outcome, body) = match tokio::time::timeout(timeout, fut).await {
         Ok(body) => {
@@ -253,7 +253,7 @@ pub(crate) async fn timed_rpc<T: Default>(
         candidate_count,
         start.elapsed().as_secs_f64() * 1000.0,
     );
-    body
+    (outcome, body)
 }
 
 pub(crate) async fn timed_results<K, V, E>(

visibility-filtering/hydration/mod.rs

@@ -11,7 +11,7 @@ pub mod viewer_hydrator;
 use crate::clients::gizmoduck_client::GizmoduckLookup;
 use crate::clients::socialgraph_client::SocialgraphClient;
 use crate::models::{
-    assemble, resolve_candidates, AuthorFeatures, AuthorId, ExclusiveContentFeatures,
+    assemble, resolve_candidates, AuthorFeatures, AuthorId, ExclusiveHydration,
     HydratedTweetCandidate, RawCandidate, SafetyLabelMap, TweetCandidateInput, TweetFeatures,
     TweetId, Viewer, ViewerAuthorRelationship, ViewerFeatures,
 };
@@ -62,14 +62,19 @@ struct CandidateFeatures {
     author_features: TweetHydrationBatch<AuthorFeatures>,
     safety_labels: HashMap<TweetId, SafetyLabelMap>,
     relationships: TweetHydrationBatch<ViewerAuthorRelationship>,
-    exclusive_content: HashMap<TweetId, Option<ExclusiveContentFeatures>>,
+    exclusive_content: HashMap<TweetId, ExclusiveHydration>,
 }
 
 impl CandidateFeatures {
     fn assemble(self, candidates: &[TweetCandidateInput]) -> Vec<HydratedTweetCandidate> {
         candidates
             .iter()
             .map(|c| {
+                let exclusive = self
+                    .exclusive_content
+                    .get(&c.tweet_id)
+                    .cloned()
+                    .unwrap_or_default();
                 assemble(
                     c,
                     self.tweet_features
@@ -82,10 +87,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.features(),
+                    exclusive.failed(),
                 )
             })
             .collect()
@@ -314,6 +317,7 @@ 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);
     }
 
     fn raw(tweet_id: u64, request_author_id: Option<u64>) -> RawCandidate {

visibility-filtering/models/exclusive_content.rs

@@ -1,5 +1,31 @@
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ExclusiveContentFeatures {
     pub conversation_author_id: u64,
     pub viewer_super_follows_author: bool,
 }
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ExclusiveHydration {
+    Public,
+    Exclusive(ExclusiveContentFeatures),
+    Failed,
+}
+
+impl Default for ExclusiveHydration {
+    fn default() -> Self {
+        ExclusiveHydration::Public
+    }
+}
+
+impl ExclusiveHydration {
+    pub fn features(&self) -> Option<ExclusiveContentFeatures> {
+        match self {
+            ExclusiveHydration::Exclusive(features) => Some(features.clone()),
+            ExclusiveHydration::Public | ExclusiveHydration::Failed => None,
+        }
+    }
+
+    pub fn failed(&self) -> bool {
+        matches!(self, ExclusiveHydration::Failed)
+    }
+}

visibility-filtering/models/mod.rs

@@ -6,7 +6,7 @@ pub mod tweet;
 pub mod viewer;
 
 pub use author::{AuthorFeatures, AuthorLabel, AuthorLabelSet};
-pub use exclusive_content::ExclusiveContentFeatures;
+pub use exclusive_content::{ExclusiveContentFeatures, ExclusiveHydration};
 pub use relationship::ViewerAuthorRelationship;
 pub use safety_labels::{SafetyLabelMap, SafetyLabelType};
 pub use tweet::{CoreFeature, MediaFeature, NsfwFeature, TweetFeatures};
@@ -74,6 +74,7 @@ pub struct HydratedTweetCandidate {
     pub safety_labels: SafetyLabelMap,
     pub relationship: ViewerAuthorRelationship,
     pub exclusive_content: Option<ExclusiveContentFeatures>,
+    pub exclusive_hydration_failed: bool,
 }
 
 impl HydratedTweetCandidate {
@@ -146,6 +147,7 @@ pub fn assemble(
     safety_labels: SafetyLabelMap,
     relationship: ViewerAuthorRelationship,
     exclusive_content: Option<ExclusiveContentFeatures>,
+    exclusive_hydration_failed: bool,
 ) -> HydratedTweetCandidate {
     HydratedTweetCandidate {
         tweet_id: candidate.tweet_id.0,
@@ -155,6 +157,7 @@ pub fn assemble(
         safety_labels,
         relationship,
         exclusive_content,
+        exclusive_hydration_failed,
     }
 }
 

visibility-filtering/rules/context.rs

@@ -191,6 +191,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)]

visibility-filtering/rules/golden_corpus.rs

@@ -163,6 +163,12 @@ fn exclusive_candidate(viewer_super_follows_author: bool) -> HydratedTweetCandid
     c
 }
 
+fn exclusive_hydration_failed_candidate() -> HydratedTweetCandidate {
+    let mut c = candidate().build();
+    c.exclusive_hydration_failed = true;
+    c
+}
+
 fn viewer_in_country(code: &str) -> ViewerFeatures {
     ViewerFeatures {
         country_code: Some(code.to_string()),
@@ -692,6 +698,22 @@ fn exclusive_content_cases() -> Vec<Case> {
             expected_action: Drop(FilteredReason::ExclusiveTweet),
             expected_decided_by: Some("DropExclusiveTweetContentRule"),
         },
+        Case {
+            name: "exclusive_hydration_failure_drops",
+            level: TimelineHome,
+            viewer: viewer(VIEWER_ID),
+            candidate: exclusive_hydration_failed_candidate(),
+            expected_action: Drop(FilteredReason::ExclusiveTweet),
+            expected_decided_by: Some("ExclusiveHydrationFailureDropRule"),
+        },
+        Case {
+            name: "exclusive_hydration_failure_drops_recommendations",
+            level: TimelineHomeRecommendations,
+            viewer: viewer(VIEWER_ID),
+            candidate: exclusive_hydration_failed_candidate(),
+            expected_action: Drop(FilteredReason::ExclusiveTweet),
+            expected_decided_by: Some("ExclusiveHydrationFailureDropRule"),
+        },
     ]
 }
 

visibility-filtering/rules/registry.rs

@@ -220,6 +220,7 @@ rust_vf:
                 "SensitiveViewerLoggedOutDropRule",
                 "SensitiveViewerUnderageDropRule",
                 "SensitiveViewerNoStatedAgeDropRule",
+                "ExclusiveHydrationFailureDropRule",
                 "DropExclusiveTweetContentRule",
                 "NsfwHighPrecisionInterstitialRule",
                 "GoreAndViolenceInterstitialRule",