From 9e5dcbb8c49216ec7af6b4c6b94058013b0f8560 Mon Sep 17 00:00:00 2001
From: rays <ray.slakinski@gmail.com>
Date: Fri, 16 Oct 2026 18:15:36 +0000
Subject: [PATCH] Signals go straight through their queue, not 25 items every 5
 minutes (#260)

unjudged() limits rows, one for each item and Signal, to 50, and the pass took the queue for
empty whenever the items numbered fewer than 50. With two Signals every batch is 25 items, so the
re-judge after #259 crawled at 25 items every 5 minutes. A batch is now full by its rows. A pass
TypeSafe stopped as malformed waits for the next wake however full it was, rather than asking again
at once.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/ipx/src/signals.rs | 24 ++++++++++++++++++++++--
 1 file changed, 22 insertions(+), 2 deletions(-)

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<bool> {
     async {
         let mut refused = HashSet::new();
         let mut hits_for = std::collections::BTreeMap::<i64, usize>::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<bool> {
                 // 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<bool> {
         }
         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<u16> {
     e.downcast_ref::<Refused>().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::<usize>() 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::<Vec<_>>()), "25 items of two Signals each is the 50 asked for");
+        assert!(!full(&(0..24).map(|_| item(2)).collect::<Vec<_>>()));
+    }
+
     #[test]
     fn each_signal_is_its_own_question_on_the_items_words() {
         let item = Unjudged {