diff --git a/crates/ipx/src/signals.rs b/crates/ipx/src/signals.rs index 47bab2f..b10a03c 100644 --- a/crates/ipx/src/signals.rs +++ b/crates/ipx/src/signals.rs @@ -53,6 +53,7 @@ async fn pass(ctx: &Ctx) -> Result { async { let mut refused = HashSet::new(); let mut hits_for = std::collections::BTreeMap::::new(); + let mut stopped = false; for item in &items { if refused.contains(&item.user_id) { continue; @@ -76,6 +77,7 @@ async fn pass(ctx: &Ctx) -> Result { // the rest wait for the next pass rather than each failing the same way (#257). Err(e) if malformed(&e) => { tracing::warn!(user = item.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signals request as malformed"); + stopped = true; break; } // Left unjudged, so it is asked again next pass. @@ -88,8 +90,9 @@ async fn pass(ctx: &Ctx) -> Result { } let some_refused = !refused.is_empty(); ctx.signals_refused.lock().unwrap().extend(refused); - // A pass that only failed would otherwise go straight round again on the same items. - anyhow::Ok(items.len() as i64 == BATCH || some_refused) + // A pass that only failed would otherwise go straight round again on the same items, and + // one TypeSafe refused as malformed would ask again at once, however full. + anyhow::Ok(!stopped && (full(&items) || some_refused)) } .instrument(span) .await @@ -115,6 +118,13 @@ fn status(e: &anyhow::Error) -> Option { e.downcast_ref::().map(|r| r.status) } +/// Whether the batch was as long as `unjudged` gives, so more may be waiting. Its limit is on +/// rows, one for each item and Signal: with two Signals, 50 is 25 items, and comparing the items +/// with 50 took every batch for the last, 25 items every 5 minutes (#260). +fn full(items: &[Unjudged]) -> bool { + items.iter().map(|i| i.signals.len()).sum::() as i64 >= BATCH +} + fn refuses_key(e: &anyhow::Error) -> bool { matches!(status(e), Some(401 | 402 | 403)) } @@ -221,6 +231,16 @@ mod tests { assert_eq!(hit(&answer, 5), None, "a question not answered is asked again"); } + #[test] + fn a_batch_is_full_by_its_rows_not_its_items() { + let item = |n: usize| Unjudged { + user_id: 1, key: "k".into(), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None, + signals: (0..n as i64).map(|id| Ask { id, name: "s".into(), keywords: vec![], description: None }).collect(), + }; + assert!(full(&(0..25).map(|_| item(2)).collect::>()), "25 items of two Signals each is the 50 asked for"); + assert!(!full(&(0..24).map(|_| item(2)).collect::>())); + } + #[test] fn each_signal_is_its_own_question_on_the_items_words() { let item = Unjudged {