Stop IneligibleSubscriptionFilter from dropping exclusive posts on a subscription-list miss

home-mixer/filters/ineligible_subscription_filter.rs

@@ -18,18 +18,37 @@ impl Filter<ScoredPostsQuery, PostCandidate> for IneligibleSubscriptionFilter {
             .map(|id| *id as u64)
             .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,
-                });
+        let (kept, removed): (Vec<_>, Vec<_>) = candidates
+            .into_iter()
+            .partition(|candidate| keep_candidate(query, candidate, &subscribed_user_ids));
 
         FilterResult { kept, removed }
     }
 }
 
+fn keep_candidate(
+    query: &ScoredPostsQuery,
+    candidate: &PostCandidate,
+    subscribed_user_ids: &HashSet<u64>,
+) -> bool {
+    let Some(author_id) = candidate.subscription_author_id else {
+        return true;
+    };
+
+    // Same exemption VF DropExclusiveTweetContentRule gives the conversation author.
+    if author_id == query.user_id {
+        return true;
+    }
+
+    // Query hydrators ignore errors, so a socialgraph miss leaves an empty list.
+    // Treat that as unknown, not "subscribed to nobody", and let VF decide.
+    if !query.subscribed_user_ids_hydrated {
+        return true;
+    }
+
+    subscribed_user_ids.contains(&author_id)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -41,12 +60,20 @@ mod tests {
         }
     }
 
+    fn query(user_id: u64, subscribed: Vec<i64>, hydrated: bool) -> ScoredPostsQuery {
+        let mut query = ScoredPostsQuery {
+            user_id,
+            subscribed_user_ids_hydrated: hydrated,
+            ..Default::default()
+        };
+        query.user_features.subscribed_user_ids = subscribed;
+        query
+    }
+
     #[tokio::test]
-    async fn keeps_only_subscribed_authors() {
+    async fn keeps_only_subscribed_authors_when_list_hydrated() {
         let filter = IneligibleSubscriptionFilter;
-        let mut query = ScoredPostsQuery::default();
-        query.user_features.subscribed_user_ids = vec![1, 2];
-
+        let query = query(99, vec![1, 2], true);
         let candidates = vec![candidate(Some(1)), candidate(Some(3)), candidate(None)];
 
         let result = filter.filter(&query, candidates);
@@ -66,4 +93,56 @@ mod tests {
             .iter()
             .any(|c| c.subscription_author_id == Some(3)));
     }
+
+    #[tokio::test]
+    async fn keeps_exclusive_posts_when_viewer_is_conversation_author() {
+        let filter = IneligibleSubscriptionFilter;
+        let query = query(7, vec![1], true);
+        let candidates = vec![candidate(Some(7)), candidate(Some(3))];
+
+        let result = filter.filter(&query, candidates);
+
+        assert_eq!(result.kept.len(), 1);
+        assert_eq!(result.kept[0].subscription_author_id, Some(7));
+        assert_eq!(result.removed.len(), 1);
+        assert_eq!(result.removed[0].subscription_author_id, Some(3));
+    }
+
+    #[tokio::test]
+    async fn keeps_exclusive_posts_when_subscription_list_was_not_hydrated() {
+        let filter = IneligibleSubscriptionFilter;
+        let query = query(99, vec![], false);
+        let candidates = vec![candidate(Some(3)), candidate(Some(4)), candidate(None)];
+
+        let result = filter.filter(&query, candidates);
+
+        assert_eq!(result.kept.len(), 3);
+        assert!(result.removed.is_empty());
+    }
+
+    #[tokio::test]
+    async fn drops_exclusive_posts_when_hydrated_list_is_empty() {
+        let filter = IneligibleSubscriptionFilter;
+        let query = query(99, vec![], true);
+        let candidates = vec![candidate(Some(3)), candidate(None)];
+
+        let result = filter.filter(&query, candidates);
+
+        assert_eq!(result.kept.len(), 1);
+        assert!(result.kept[0].subscription_author_id.is_none());
+        assert_eq!(result.removed.len(), 1);
+        assert_eq!(result.removed[0].subscription_author_id, Some(3));
+    }
+
+    #[tokio::test]
+    async fn conversation_author_kept_even_when_list_missing() {
+        let filter = IneligibleSubscriptionFilter;
+        let query = query(7, vec![], false);
+        let candidates = vec![candidate(Some(7))];
+
+        let result = filter.filter(&query, candidates);
+
+        assert_eq!(result.kept.len(), 1);
+        assert!(result.removed.is_empty());
+    }
 }

home-mixer/models/query.rs

@@ -119,6 +119,9 @@ pub struct ScoredPostsQuery {
     pub impressed_post_ids: Vec<u64>,
     pub push_to_home_post_id: Option<u64>,
     pub seed_candidate_post_ids: Vec<u64>,
+    /// Set only after SubscribedUserIdsQueryHydrator succeeds.
+    /// An empty subscribed list with this false is a hydrator miss, not "no subscriptions".
+    pub subscribed_user_ids_hydrated: bool,
     #[serde(serialize_with = "serialize_debug")]
     pub following_pagination_meta: Arc<OnceLock<FollowingPaginationMeta>>,
 }
@@ -227,6 +230,7 @@ impl ScoredPostsQuery {
             impressed_post_ids: Vec::new(),
             push_to_home_post_id,
             seed_candidate_post_ids: Vec::new(),
+            subscribed_user_ids_hydrated: false,
             following_pagination_meta: Arc::new(OnceLock::new()),
         }
     }

home-mixer/query_hydrators/subscribed_user_ids_query_hydrator.rs

@@ -23,11 +23,123 @@ impl QueryHydrator<ScoredPostsQuery> for SubscribedUserIdsQueryHydrator {
                 subscribed_user_ids,
                 ..Default::default()
             },
+            subscribed_user_ids_hydrated: true,
             ..Default::default()
         })
     }
 
     fn update(&self, query: &mut ScoredPostsQuery, hydrated: ScoredPostsQuery) {
         query.user_features.subscribed_user_ids = hydrated.user_features.subscribed_user_ids;
+        query.subscribed_user_ids_hydrated = hydrated.subscribed_user_ids_hydrated;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::models::user_features::UserFeatures;
+    use std::collections::HashSet;
+    use tonic::Status;
+
+    struct MockSocialGraph {
+        ids: Result<Vec<i64>, Status>,
+    }
+
+    #[async_trait]
+    impl SocialGraphClientOps for MockSocialGraph {
+        async fn get_following_list(&self, _user_id: u64) -> Result<Vec<u64>, Status> {
+            Ok(vec![])
+        }
+        async fn check_blocked_by(
+            &self,
+            _viewer_id: u64,
+            _author_ids: &[u64],
+        ) -> Result<HashSet<u64>, Status> {
+            Ok(HashSet::new())
+        }
+        async fn check_followed_by(
+            &self,
+            _viewer_id: u64,
+            _user_ids: &[u64],
+        ) -> Result<HashSet<u64>, Status> {
+            Ok(HashSet::new())
+        }
+        async fn get_blocked_user_ids(&self, _viewer_id: u64) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+        async fn get_muted_user_ids(&self, _viewer_id: u64) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+        async fn get_followed_user_ids(&self, _viewer_id: u64) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+        async fn get_follower_ids(&self, _user_id: u64) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+        async fn get_subscribed_user_ids(&self, _viewer_id: u64) -> Result<Vec<i64>, Status> {
+            match &self.ids {
+                Ok(ids) => Ok(ids.clone()),
+                Err(status) => Err(Status::new(status.code(), status.message())),
+            }
+        }
+        async fn get_device_following_user_ids(&self, _viewer_id: u64) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+        async fn get_hide_recommendations_user_ids(
+            &self,
+            _viewer_id: u64,
+        ) -> Result<Vec<i64>, Status> {
+            Ok(vec![])
+        }
+    }
+
+    fn hydrator(ids: Result<Vec<i64>, Status>) -> SubscribedUserIdsQueryHydrator {
+        SubscribedUserIdsQueryHydrator {
+            socialgraph_client: Arc::new(MockSocialGraph { ids }),
+        }
+    }
+
+    #[tokio::test]
+    async fn successful_hydrate_marks_list_ready() {
+        let hydrator = hydrator(Ok(vec![10, 20]));
+        let result = hydrator.hydrate(&ScoredPostsQuery::default()).await.unwrap();
+        assert_eq!(result.user_features.subscribed_user_ids, vec![10, 20]);
+        assert!(result.subscribed_user_ids_hydrated);
+    }
+
+    #[tokio::test]
+    async fn successful_empty_list_is_still_hydrated() {
+        let hydrator = hydrator(Ok(vec![]));
+        let result = hydrator.hydrate(&ScoredPostsQuery::default()).await.unwrap();
+        assert!(result.user_features.subscribed_user_ids.is_empty());
+        assert!(result.subscribed_user_ids_hydrated);
+    }
+
+    #[tokio::test]
+    async fn socialgraph_error_does_not_mark_list_ready() {
+        let hydrator = hydrator(Err(Status::unavailable("sg down")));
+        assert!(hydrator.hydrate(&ScoredPostsQuery::default()).await.is_err());
+    }
+
+    #[test]
+    fn update_copies_ids_and_hydrated_flag() {
+        let hydrator = hydrator(Ok(vec![]));
+        let mut query = ScoredPostsQuery::default();
+        assert!(!query.subscribed_user_ids_hydrated);
+
+        hydrator.update(
+            &mut query,
+            ScoredPostsQuery {
+                user_features: UserFeatures {
+                    subscribed_user_ids: vec![1, 2],
+                    ..Default::default()
+                },
+                subscribed_user_ids_hydrated: true,
+                ..Default::default()
+            },
+        );
+
+        assert_eq!(query.user_features.subscribed_user_ids, vec![1, 2]);
+        assert!(query.subscribed_user_ids_hydrated);
     }
 }