diff --git a/phoenix-rankall-strato/columns/phoenix_rank_all/getPostPhoenixRankAllMetadata.strato b/phoenix-rankall-strato/columns/phoenix_rank_all/getPostPhoenixRankAllMetadata.strato index 8a79f319..77207a05 100644 --- a/phoenix-rankall-strato/columns/phoenix_rank_all/getPostPhoenixRankAllMetadata.strato +++ b/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) } diff --git a/phoenix-rankall-strato/columns/phoenix_rank_all/phoenixRankAllCandidateProcessor.strato b/phoenix-rankall-strato/columns/phoenix_rank_all/phoenixRankAllCandidateProcessor.strato index 552e4fd2..82d9476e 100644 --- a/phoenix-rankall-strato/columns/phoenix_rank_all/phoenixRankAllCandidateProcessor.strato +++ b/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 } #.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 => diff --git a/phoenix-rankall-strato/lib/eventProcessing.strato b/phoenix-rankall-strato/lib/eventProcessing.strato index 51d7db24..29e83c7b 100644 --- a/phoenix-rankall-strato/lib/eventProcessing.strato +++ b/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.fetch(postId, ()).v, #Tweet.fetch(postId, ()).v, - #Tweet.fetch(postId, ()).v + #Tweet.fetch(postId, ()).v, + #Tweet.fetch(postId, ()).v, + #Tweet.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 = { # .fetch(postId, { safetyLevel = TimelineHomeRecommendations }).v } catch { - case _ => None + case _ => Some(true) } - verdictOpt.getOrElse(false) + verdictOpt.getOrElse(true) } else { - #Tweet - .fetch(postId, { safetyLevel = TimelineHomeRecommendations, forUserId = None }).v - .getOrElse(false) + val verdictOpt = try { + #Tweet + .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, diff --git a/phoenix-rankall/src/processor/main_processor.rs b/phoenix-rankall/src/processor/main_processor.rs index 470a0290..b2fa2611 100644 --- a/phoenix-rankall/src/processor/main_processor.rs +++ b/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(); diff --git a/phoenix-rankall/src/processor/metadata_processor.rs b/phoenix-rankall/src/processor/metadata_processor.rs index 89e71728..c2b91c9e 100644 --- a/phoenix-rankall/src/processor/metadata_processor.rs +++ b/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"); diff --git a/phoenix-rankall/src/processor/mod.rs b/phoenix-rankall/src/processor/mod.rs index 9e9e90e6..1bca998d 100644 --- a/phoenix-rankall/src/processor/mod.rs +++ b/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)); + } +} diff --git a/phoenix-rankall/src/processor/sid_processor.rs b/phoenix-rankall/src/processor/sid_processor.rs index 3109b0d1..8a69d2ae 100644 --- a/phoenix-rankall/src/processor/sid_processor.rs +++ b/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; diff --git a/phoenix-rankall/src/processor/sid_tail_processor.rs b/phoenix-rankall/src/processor/sid_tail_processor.rs index 4c7c10cc..c7aa9afa 100644 --- a/phoenix-rankall/src/processor/sid_tail_processor.rs +++ b/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); diff --git a/phoenix-rankall/src/processor/topic_processor.rs b/phoenix-rankall/src/processor/topic_processor.rs index 96483096..31754e3d 100644 --- a/phoenix-rankall/src/processor/topic_processor.rs +++ b/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();