YouTube feeds are read one at a time, two seconds apart (#386)

A scan fetches six feeds ahead of its turn, so channels next to each
other alphabetically (android-authority-2, android-developers, apple,
architectural-digest-2) went to YouTube together and failed together,
while one asked alone a minute later read. feed::fetch now takes a turn
before each request for a YouTube feed: one at a time per process, two
seconds after the last, its wait a `youtube_turn` span. The retries
take their turn too, in place of their one-second sleep. The directory
builds the same fetch and paces its own checks the same way.

39 channels due together add up to 80 s to that scan.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/feed/src/fetch.rs

@@ -41,9 +41,26 @@ pub static FETCHED_BYTES: AtomicU64 = AtomicU64::new(0);
 /// catalogue was 62 MB, and 50 MB here stopped it updating (#357).
 pub(crate) const MAX_BODY: usize = 200 << 20;
 
-/// How many times a YouTube feed's spurious failure is asked again, and how long between.
+/// How many times a YouTube feed's spurious failure is asked again, each after its turn.
 const YOUTUBE_RETRIES: u32 = 2;
-const YOUTUBE_RETRY_AFTER: Duration = Duration::from_secs(1);
+
+/// The least time between two requests for YouTube feeds from this process. A scan asked for six
+/// at once, and channels in a row failed together while one asked alone a minute later read
+/// (#386); asked again a second later, they failed again.
+const YOUTUBE_GAP: Duration = Duration::from_secs(2);
+static YOUTUBE_NEXT: tokio::sync::Mutex<Option<tokio::time::Instant>> = tokio::sync::Mutex::const_new(None);
+
+/// Waits until `gap` after the last request's turn, one caller at a time: tokio's lock is fair,
+/// so waiting feeds go in the order they asked. shortcut: 39 channels due together add up to 80 s
+/// to that scan; a scan's other feeds wait behind them only when they come after in the order.
+#[tracing::instrument(name = "youtube_turn", skip_all)]
+async fn take_turn(next: &tokio::sync::Mutex<Option<tokio::time::Instant>>, gap: Duration) {
+    let mut next = next.lock().await;
+    if let Some(at) = *next {
+        tokio::time::sleep_until(at).await;
+    }
+    *next = Some(tokio::time::Instant::now() + gap);
+}
 
 /// YouTube answers some feed requests with a 404 or 500 it does not mean, in tens of
 /// milliseconds, while the next request for the same feed reads; checked once a day, a channel
@@ -89,6 +106,9 @@ pub async fn fetch(
     let mut retries = YOUTUBE_RETRIES;
     // Ten redirects, and the retries beside them.
     for _ in 0..10 + YOUTUBE_RETRIES {
+        if is_youtube_feed(url.as_str()) {
+            take_turn(&YOUTUBE_NEXT, YOUTUBE_GAP).await;
+        }
         let mut req = client.get(url.clone()).timeout(FEED_TIMEOUT);
         if let Some(tag) = etag {
             req = req.header(IF_NONE_MATCH, tag);
@@ -122,7 +142,6 @@ pub async fn fetch(
         }
         if retries > 0 && youtube_spurious(&url, status) {
             retries -= 1;
-            tokio::time::sleep(YOUTUBE_RETRY_AFTER).await;
             continue;
         }
         if !status.is_success() {
@@ -192,6 +211,25 @@ mod tests {
         format!("http://{addr}")
     }
 
+    #[tokio::test]
+    async fn youtube_turns_are_a_gap_apart_one_at_a_time() {
+        let (next, gap) = (tokio::sync::Mutex::const_new(None), std::time::Duration::from_millis(50));
+        let start = tokio::time::Instant::now();
+        // The first goes at once; three asking together then go one gap after another.
+        super::take_turn(&next, gap).await;
+        assert!(start.elapsed() < gap);
+        let t = || async {
+            super::take_turn(&next, gap).await;
+            start.elapsed()
+        };
+        let (a, b, c) = tokio::join!(t(), t(), t());
+        let mut at = [a, b, c];
+        at.sort();
+        for (i, at) in at.iter().enumerate() {
+            assert!(*at >= gap * (i as u32 + 1), "turn {} at {at:?}", i + 1);
+        }
+    }
+
     #[test]
     fn only_a_youtube_feeds_404_or_5xx_is_asked_again() {
         use reqwest::StatusCode;