Stop rank-all from indexing NSFW and VF-dropped posts into SFW retrieval

phoenix-rankall-strato/columns/phoenix_rank_all/getPostPhoenixRankAllMetadata.strato

@@ -13,20 +13,25 @@ val executeOp = Op.execute({ idempotent = true })[Long, Option[PhoenixRankAllCan
   postMetadata match {
     case Some(post) =>
       val authorId: Option[Long] = eventProcessing.getAuthorId(post)
-      val unifiedPostAnnotations = eventProcessing.getUnifiedPostAnnotations(postId)
-      val engagementCount = eventProcessing.getEngagementCount(postId)
-      val authorFollowers = eventProcessing.getAuthorFollowers(post)
-      val hasImage = tweetUtil.hasImage(post)
-      val hasVideo = tweetUtil.hasVideo(post)
-      Ok(Some({
-        postId = postId,
-        authorId = authorId.getOrElse(-1),
-        hasVideo = hasVideo,
-        hasImage = hasImage,
-        authorFollowersCount = authorFollowers,
-        engagementCount = engagementCount,
-        postAnnotations = unifiedPostAnnotations
-      }))
+      authorId match {
+        case Some(id) if id > 0 =>
+          val unifiedPostAnnotations = eventProcessing.getUnifiedPostAnnotations(postId)
+          val engagementCount = eventProcessing.getEngagementCount(postId)
+          val authorFollowers = eventProcessing.getAuthorFollowers(post)
+          val hasImage = tweetUtil.hasImage(post)
+          val hasVideo = tweetUtil.hasVideo(post)
+          Ok(Some({
+            postId = postId,
+            authorId = id,
+            hasVideo = hasVideo,
+            hasImage = hasImage,
+            authorFollowersCount = authorFollowers,
+            engagementCount = engagementCount,
+            postAnnotations = unifiedPostAnnotations
+          }))
+        case _ =>
+          Ok(None)
+      }
     case _ =>
       Ok(None)
   }

phoenix-rankall-strato/columns/phoenix_rank_all/phoenixRankAllCandidateProcessor.strato

@@ -329,7 +329,11 @@ def buildImagineIndex(candidate: PhoenixRankAllCandidate, stats: Stats.StatsRece
   }
 }
 
-def buildMetadataDump(candidate: PhoenixRankAllCandidate, stats: Stats.StatsReceiver): Unit = {
+def buildMetadataDump(
+  candidate: PhoenixRankAllCandidate,
+  stats: Stats.StatsReceiver,
+  indexName: Option[String]
+): Unit = {
   candidate.tweetMedadata match {
     case Some(tweet) =>
       val hasImage = tweetUtil.hasImage(tweet)
@@ -343,6 +347,7 @@ def buildMetadataDump(candidate: PhoenixRankAllCandidate, stats: Stats.StatsRece
         authorFollowersCount = candidate.authorFollowers,
         engagementCount = candidate.engagementCount,
         videoDurationMs = videoDurationMsOpt,
+        indexName = indexName
       }
       #<hydra/kafka/metadataEventKafkaTopic>.insert((), flatObject)
       stats.counter("metadata_dump_created").incr(1)
@@ -406,7 +411,7 @@ val executeOp = Op.execute({ idempotent = true })[PhoenixRankAllIndexingRequest,
     case Some(post) =>
       val authorId: Option[Long] = eventProcessing.getAuthorId(post)
       authorId match {
-        case Some(id) =>
+        case Some(id) if id > 0 =>
           val unifiedPostAnnotations = eventProcessing.getUnifiedPostAnnotations(postId)
           val engagementCount = eventProcessing.getEngagementCount(postId)
           val authorFollowers = eventProcessing.getAuthorFollowers(post)
@@ -420,16 +425,24 @@ val executeOp = Op.execute({ idempotent = true })[PhoenixRankAllIndexingRequest,
             authorFollowers = authorFollowers
           }
           val hasImmersiveVideo = eventProcessing.hasImmersiveVideo(post)
+          val isNsfwAuthor = !eventProcessing.isAuthorEligible(post)
+          val isAdultPost = eventProcessing.isAdultPost(unifiedPostAnnotations)
+          val shouldDropPostByVF = eventProcessing.shouldDropPostByVF(postId)
+          val shouldDropPost = shouldDropPostByVF || isAdultPost || isNsfwAuthor
 
           if (eventSource == EvergreenVideo) {
-            if(hasImmersiveVideo) {
+            if (hasImmersiveVideo && !shouldDropPost) {
               buildEvergreenVideoIndex(candidate, receiverStats)
               numCandidatesProcessed.incr(1)
+            } else if (shouldDropPost) {
+              numSkippedPosts.incr(1)
             }
           } else if (eventSource == EvergreenNsfwVideo) {
-            if(hasImmersiveVideo) {
+            if (hasImmersiveVideo && !shouldDropPostByVF) {
               buildEvergreenNsfwVideoIndex(candidate, receiverStats)
               numCandidatesProcessed.incr(1)
+            } else if (shouldDropPostByVF) {
+              numSkippedPosts.incr(1)
             }
           } else {
             val isCommunityPost = eventProcessing.isCommunityPost(post)
@@ -440,9 +453,6 @@ val executeOp = Op.execute({ idempotent = true })[PhoenixRankAllIndexingRequest,
             }
 
             val isReply = eventProcessing.isReply(post)
-            val isAdultPost = eventProcessing.isNsfwPost(unifiedPostAnnotations)
-            val shouldDropPostByVF = eventProcessing.shouldDropPostByVF(postId)
-            val shouldDropPost = shouldDropPostByVF || isAdultPost
             if(isCommunityPost) {
               numSkippedCommunity.incr(1)
             } else if (isReply) {
@@ -452,7 +462,7 @@ val executeOp = Op.execute({ idempotent = true })[PhoenixRankAllIndexingRequest,
             } else if (shouldDropPost) {
               if (hasImmersiveVideo && eventSource == Fav && isAdultPost) {
                 buildNsfwVideoIndex(candidate, receiverStats)
-                buildMetadataDump(candidate, receiverStats)
+                buildMetadataDump(candidate, receiverStats, Some("nsfw_metadata"))
                 buildMMEmbMetadataDump(candidate, receiverStats)
               } else {
                 numSkippedPosts.incr(1)
@@ -469,21 +479,21 @@ val executeOp = Op.execute({ idempotent = true })[PhoenixRankAllIndexingRequest,
                     build1FavVideoIndex(candidate, receiverStats)
                     buildImagineIndex(candidate, receiverStats)
                   }
-                  buildMetadataDump(candidate, receiverStats)
+                  buildMetadataDump(candidate, receiverStats, Some("metadata"))
                   buildMMEmbMetadataDump(candidate, receiverStats)
                 case PostCreation =>
                   buildPostCreationIndex(candidate, receiverStats)
                   if(hasImmersiveVideo) {
                     buildVideoIndex(candidate, receiverStats)
                   }
-                  buildMetadataDump(candidate, receiverStats)
+                  buildMetadataDump(candidate, receiverStats, Some("metadata"))
                 case _ =>
                   numInvalidEventSource.incr(1)
               }
               numCandidatesProcessed.incr(1)
             }
           }
-        case None =>
+        case _ =>
           numMissingAuthorId.incr(1)
       }
     case None =>

phoenix-rankall-strato/lib/eventProcessing.strato

@@ -157,7 +157,7 @@ def hasVideo(mediaEntity: com.twitter.tweetypie.MediaEntity) : Boolean = {
 def isAuthorEligible(tweet: com.twitter.tweetypie.Tweet) : Boolean = {
   tweet.coreData.map {
     coreData => (coreData.nsfwUser == false) && (coreData.nsfwAdmin == false)
-  }.getOrElse(true)
+  }.getOrElse(false)
 }
 
 def isAuthorVisible(tweet: com.twitter.tweetypie.Tweet): Boolean = {
@@ -205,10 +205,12 @@ def getTweetypieResponse(tweetId: Long): TweetypieResponse = {
 }
 
 def getPostMetadataLightweight(postId: Long): Option[Tweet] = {
-  val (pureCoreDataOpt, mediaEntitiesOpt, communitiesOpt) = (
+  val (pureCoreDataOpt, mediaEntitiesOpt, communitiesOpt, nsfwUserOpt, nsfwAdminOpt) = (
     #Tweet<tweetypie/federated/pureCoreData>.fetch(postId, ()).v,
     #Tweet<tweetypie/federated/mediaEntities>.fetch(postId, ()).v,
-    #Tweet<tweetypie/federated/communities>.fetch(postId, ()).v
+    #Tweet<tweetypie/federated/communities>.fetch(postId, ()).v,
+    #Tweet<tweetypie/federated/nsfwUser>.fetch(postId, ()).v,
+    #Tweet<tweetypie/federated/nsfwAdmin>.fetch(postId, ()).v
   )
 
   pureCoreDataOpt.map { data =>
@@ -221,6 +223,8 @@ def getPostMetadataLightweight(postId: Long): Option[Tweet] = {
         createdVia = data.createdVia,
         reply = data.reply,
         share = data.share,
+        nsfwUser = nsfwUserOpt.getOrElse(false),
+        nsfwAdmin = nsfwAdminOpt.getOrElse(false),
       }: com.twitter.tweetypie.TweetCoreData),
       media = mediaEntitiesOpt,
       communities = communitiesOpt,
@@ -254,13 +258,17 @@ def shouldDropPostByVF(postId: Long): Boolean = {
       #<visibility/xai/shouldDropTweet>
         .fetch(postId, { safetyLevel = TimelineHomeRecommendations }).v
     } catch {
-      case _ => None
+      case _ => Some(true)
     }
-    verdictOpt.getOrElse(false)
+    verdictOpt.getOrElse(true)
   } else {
-    #Tweet<visibility/service/shouldDropTweetV2>
-      .fetch(postId, { safetyLevel = TimelineHomeRecommendations, forUserId = None }).v
-      .getOrElse(false)
+    val verdictOpt = try {
+      #Tweet<visibility/service/shouldDropTweetV2>
+        .fetch(postId, { safetyLevel = TimelineHomeRecommendations, forUserId = None }).v
+    } catch {
+      case _ => Some(true)
+    }
+    verdictOpt.getOrElse(true)
   }
 }
 
@@ -645,6 +653,7 @@ Library({
     isNsfwPost = isNsfwPost,
     isAuthorVisible = isAuthorVisible,
     isRepost = isRepost,
+    isAuthorEligible = isAuthorEligible,
     hasImmersiveVideo = hasImmersiveVideo,
     hasValidImmersiveVideo = hasValidImmersiveVideo,
     isImaginePost = isImaginePost,

phoenix-rankall/src/processor/main_processor.rs

@@ -49,7 +49,7 @@ impl RecordProcessor for MainProcessor {
             let author_id = obj.author_id.unwrap_or(0);
             let index_name = obj.index_name.unwrap_or_default();
 
-            if post_id == 0 || author_id == 0 || index_name.is_empty() {
+            if !super::valid_index_ids(post_id, author_id) || index_name.is_empty() {
                 self.stats.total_invalid += 1;
                 continue;
             }
@@ -130,6 +130,22 @@ mod tests {
         assert_eq!(proc.stats().total_success, 1);
     }
 
+    #[test]
+    fn skip_records_with_sentinel_author_id() {
+        let mut proc = MainProcessor::new();
+        let raw = vec![
+            make_thrift_bytes(100, -1, "1fav"),
+            make_thrift_bytes(200, -1, "video"),
+            make_thrift_bytes(300, 30, "1fav"),
+        ];
+
+        let results = proc.process_batch(&raw);
+        assert_eq!(results.len(), 1, "sentinel author_id=-1 must not be indexed");
+        assert_eq!(results[0].post_id(), 300);
+        assert_eq!(proc.stats().total_invalid, 2);
+        assert_eq!(proc.stats().total_success, 1);
+    }
+
     #[test]
     fn handle_corrupt_payload() {
         let mut proc = MainProcessor::new();

phoenix-rankall/src/processor/metadata_processor.rs

@@ -44,7 +44,7 @@ impl RecordProcessor for MetadataProcessor {
             let post_id = obj.post_id.unwrap_or(0);
             let author_id = obj.author_id.unwrap_or(0);
 
-            if post_id == 0 || author_id == 0 {
+            if !super::valid_index_ids(post_id, author_id) {
                 self.stats.total_invalid += 1;
                 continue;
             }
@@ -185,6 +185,20 @@ mod tests {
         assert_eq!(proc.stats().total_invalid, 1);
     }
 
+    #[test]
+    fn skips_sentinel_author_id() {
+        let mut proc = MetadataProcessor::new("metadata");
+        let raw = vec![
+            make_metadata_bytes(100, -1, "metadata", false, 0),
+            make_metadata_bytes(200, 20, "metadata", false, 0),
+        ];
+
+        let results = proc.process_batch(&raw);
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].post_id(), 200);
+        assert_eq!(proc.stats().total_invalid, 1);
+    }
+
     #[test]
     fn defaults_missing_engagement() {
         let mut proc = MetadataProcessor::new("metadata");

phoenix-rankall/src/processor/mod.rs

@@ -53,3 +53,21 @@ pub trait RecordProcessor: Send + Sync {
 
     fn stats(&self) -> &ProcessorStats;
 }
+
+pub fn valid_index_ids(post_id: i64, author_id: i64) -> bool {
+    post_id > 0 && author_id > 0
+}
+
+#[cfg(test)]
+mod tests {
+    use super::valid_index_ids;
+
+    #[test]
+    fn rejects_zero_and_sentinel_ids() {
+        assert!(!valid_index_ids(0, 10));
+        assert!(!valid_index_ids(100, 0));
+        assert!(!valid_index_ids(100, -1));
+        assert!(!valid_index_ids(-5, 10));
+        assert!(valid_index_ids(100, 10));
+    }
+}

phoenix-rankall/src/processor/sid_processor.rs

@@ -60,7 +60,7 @@ impl RecordProcessor for SidProcessor {
             let author_id = obj.author_id.unwrap_or(0);
             let index_name = obj.index_name.unwrap_or_default();
 
-            if post_id == 0 || author_id == 0 || index_name.is_empty() {
+            if !super::valid_index_ids(post_id, author_id) || index_name.is_empty() {
                 self.stats.total_invalid += 1;
                 continue;
             }
@@ -174,6 +174,7 @@ mod tests {
         let raw = vec![
             make_thrift_bytes(0, 10, "1fav"),
             make_thrift_bytes(100, 0, "1fav"),
+            make_thrift_bytes(100, -1, "1fav"),
             make_thrift_bytes(100, 10, ""),
         ];
         let results = process_in_blocking(move || proc.process_batch(&raw)).await;

phoenix-rankall/src/processor/sid_tail_processor.rs

@@ -63,7 +63,7 @@ impl RecordProcessor for SidTailProcessor {
 
             let post_id = obj.post_id.unwrap_or(0);
             let author_id = obj.author_id.unwrap_or(0);
-            if post_id == 0 || author_id == 0 {
+            if !super::valid_index_ids(post_id, author_id) {
                 self.stats.total_invalid += 1;
                 continue;
             }
@@ -216,6 +216,19 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn skips_nsfw_metadata_and_sentinel_author() {
+        let mut proc = make_processor(1000, 0);
+        let batch = vec![
+            make_bytes(1, 10, Some("nsfw_metadata"), 10, Some(5)),
+            make_bytes(2, -1, Some("metadata"), 10, Some(5)),
+            make_bytes(3, 30, Some("metadata"), 10, Some(5)),
+        ];
+        let out = process_in_blocking(move || proc.process_batch(&batch)).await;
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].post_id(), 3);
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn min_fav_filter() {
         let mut proc = make_processor(1000, 1);

phoenix-rankall/src/processor/topic_processor.rs

@@ -48,7 +48,7 @@ impl RecordProcessor for TopicProcessor {
             let author_id = obj.author_id.unwrap_or(0);
             let index_name = obj.index_name.unwrap_or_default();
 
-            if post_id == 0 || author_id == 0 || index_name.is_empty() {
+            if !super::valid_index_ids(post_id, author_id) || index_name.is_empty() {
                 self.stats.total_invalid += 1;
                 continue;
             }
@@ -151,6 +151,20 @@ mod tests {
         }
     }
 
+    #[test]
+    fn skip_sentinel_author_id() {
+        let mut proc = TopicProcessor::new(Blacklist::default());
+        let raw = vec![
+            make_thrift_bytes(100, -1, "1fav_topic", None),
+            make_thrift_bytes(200, 20, "1fav_topic", None),
+        ];
+
+        let results = proc.process_batch(&raw);
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].post_id(), 200);
+        assert_eq!(proc.stats().total_invalid, 1);
+    }
+
     #[test]
     fn blacklist_filters_blocked_posts() {
         let mut bl = Blacklist::default();