From 32529fa24f9bb6d326ccd80b90fe886bb2a15582 Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Tue, 8 Sep 2026 19:07:43 -0400 Subject: [PATCH] ranking: fail closed when engagement-count store errors Dragonfly Err was coerced to an empty map, then Ok(default) counts. CachedHydrator cached those zeros for 60s and Phoenix unwrap_or(0) ranked the slate as if nobody engaged. --- .../engagement_counts_hydrator.rs | 143 +++++++++++++++++- 1 file changed, 135 insertions(+), 8 deletions(-) diff --git a/home-mixer/candidate_hydrators/engagement_counts_hydrator.rs b/home-mixer/candidate_hydrators/engagement_counts_hydrator.rs index e74280d8..1882ece2 100644 --- a/home-mixer/candidate_hydrators/engagement_counts_hydrator.rs +++ b/home-mixer/candidate_hydrators/engagement_counts_hydrator.rs @@ -131,16 +131,30 @@ impl CachedHydrator for EngagementCountsHydrato unique_ids.sort_unstable(); unique_ids.dedup(); + // Store errors must not become Ok(empty). CachedHydrator caches Ok, and + // as_tweet_info unwrap_or(0) would stamp Phoenix / Thompson as zero + // engagement for 60s. LanguageCode / MediaInfo already return Err here. let counts = if unique_ids.is_empty() { - HashMap::new() + Ok(HashMap::new()) } else { - self.client - .get_engagement_counts(&unique_ids) - .await - .unwrap_or_else(|e| { - warn!(error = %e, "engagement_counts_hydration dragonfly_error"); - HashMap::new() - }) + self.client.get_engagement_counts(&unique_ids).await + }; + + let counts = match counts { + Ok(counts) => counts, + Err(e) => { + warn!(error = %e, "engagement_counts_hydration dragonfly_error"); + return candidates + .iter() + .map(|c| { + if !fetch_counts(c) { + Ok(preserve_counts(c)) + } else { + Err(format!("engagement_counts dragonfly_error: {e}")) + } + }) + .collect(); + } }; candidates @@ -327,4 +341,117 @@ mod tests { assert_eq!(second[0].as_ref().unwrap().view_count, Some(5)); assert_eq!(client.calls.load(Ordering::SeqCst), 1); } + + #[derive(Default)] + struct FailingClient { + calls: AtomicUsize, + } + + #[tonic::async_trait] + impl EngagementCountsClient for FailingClient { + async fn get_engagement_counts( + &self, + _tweet_ids: &[u64], + ) -> Result, String> { + self.calls.fetch_add(1, Ordering::SeqCst); + Err("dragonfly down".to_string()) + } + } + + #[tokio::test] + async fn store_error_returns_err_instead_of_zero_counts() { + let h = EngagementCountsHydrator::new(Arc::new(FailingClient::default())).await; + let candidates = vec![PostCandidate { + tweet_id: 10, + author_id: 1, + fav_count: Some(99), + view_count: Some(50), + view_count_on_home: Some(40), + ..Default::default() + }]; + let q = query(false, &[(COUNTS, "true")]); + let result = h.hydrate_from_client(&q, &candidates).await; + assert!( + result[0] + .as_ref() + .err() + .is_some_and(|e| e.contains("dragonfly_error")), + "store error must not become Ok(zero counts), got {:?}", + result[0] + ); + } + + #[tokio::test] + async fn store_error_does_not_overwrite_or_cache_zeros() { + let client = Arc::new(FailingClient::default()); + let h = EngagementCountsHydrator::new(client.clone()).await; + let candidates = vec![PostCandidate { + tweet_id: 10, + author_id: 1, + fav_count: Some(99), + view_count: Some(50), + view_count_on_home: Some(40), + ..Default::default() + }]; + let q = query(false, &[(COUNTS, "true")]); + + let first = xai_candidate_pipeline::hydrator::Hydrator::hydrate(&h, &q, &candidates).await; + let second = xai_candidate_pipeline::hydrator::Hydrator::hydrate(&h, &q, &candidates).await; + assert!(first[0].is_err()); + assert!(second[0].is_err()); + assert_eq!( + client.calls.load(Ordering::SeqCst), + 2, + "Err must not be cached as empty counts" + ); + + let mut live = candidates; + xai_candidate_pipeline::hydrator::Hydrator::update_all(&h, &mut live, first); + assert_eq!(live[0].fav_count, Some(99)); + assert_eq!(live[0].view_count, Some(50)); + assert_eq!(live[0].view_count_on_home, Some(40)); + } + + #[tokio::test] + async fn store_error_preserves_cached_ineligible_and_errs_eligible() { + let h = EngagementCountsHydrator::new(Arc::new(FailingClient::default())).await; + let candidates = vec![ + PostCandidate { + tweet_id: 10, + author_id: 1, + author_followers_count: Some(100), + view_count: Some(5), + ..Default::default() + }, + PostCandidate { + tweet_id: 20, + author_id: 2, + author_followers_count: Some(5000), + view_count: Some(999), + ..Default::default() + }, + ]; + let q = query( + true, + &[(COLD_START, "true"), (COUNTS, "true"), (CAP, "1000")], + ); + let result = h.hydrate_from_client(&q, &candidates).await; + assert!(result[0].is_err()); + assert_eq!(result[1].as_ref().unwrap().view_count, Some(999)); + } + + #[tokio::test] + async fn successful_miss_still_hydrates_empty_counts() { + let h = hydrator(HashMap::new()).await; + let candidates = vec![PostCandidate { + tweet_id: 10, + author_id: 1, + fav_count: Some(99), + ..Default::default() + }]; + let q = query(false, &[(COUNTS, "true")]); + let result = h.hydrate_from_client(&q, &candidates).await; + assert_eq!(result[0].as_ref().unwrap().fav_count, None); + assert_eq!(result[0].as_ref().unwrap().view_count, None); + } }