From 0cbc508bf721575d1b632934abb8e24150a758a1 Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Mon, 26 Oct 2026 20:54:38 +0100
Subject: warn about dropped blob requests

---
 src/swarm/blob.rs | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

(limited to 'src/swarm')

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index d68a78c..372d964 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -51,6 +51,9 @@ const MAX_CHUNK: u32 = 4 << 20;
 /// Requests being answered at once; more are dropped (the requester asks again, or
 /// another peer). Bounds the tasks anyone can make a node spawn.
 const MAX_ANSWERING: usize = 32;
+/// At most one warning about dropped requests per this interval: a few in a burst are
+/// normal (many nodes fetching a new epoch's snapshot), a steady stream is not.
+const DROPPED_LOG_INTERVAL: Duration = Duration::from_secs(60);
 /// How long to wait for a manifest from one peer.
 const MANIFEST_TIMEOUT: Duration = Duration::from_secs(15);
 /// How long to wait for a chunk before asking another peer (raised by the peer's
@@ -67,6 +70,8 @@ pub(crate) struct Blobs {
     serving: Arc<Semaphore>,
     /// Requests being answered (`MAX_ANSWERING`).
     answering: Arc<Semaphore>,
+    /// Requests dropped since the last warning about them, and when that was.
+    dropped: Mutex<(u64, Option<Instant>)>,
     /// The blob served last.
     cache: Mutex<Option<(BlobId, Arc<Prepared>)>>,
     /// Answers for downloads in progress, by blob.
@@ -84,6 +89,7 @@ impl Blobs {
             chunk_size: chunk_size.max(1),
             serving: Arc::new(Semaphore::new(MAX_SERVING)),
             answering: Arc::new(Semaphore::new(MAX_ANSWERING)),
+            dropped: Mutex::new((0, None)),
             cache: Mutex::new(None),
             fetches: Mutex::new(HashMap::new()),
         }
@@ -95,6 +101,21 @@ impl Blobs {
         self.fetches.lock_or_recover()
     }
 
+    /// Records a dropped request. Returns how many were dropped since the last warning
+    /// (this one included) if one is due now.
+    fn note_dropped(&self) -> Option<u64> {
+        let mut dropped = self.dropped.lock_or_recover();
+        dropped.0 += 1;
+        let due = dropped
+            .1
+            .is_none_or(|last| last.elapsed() >= DROPPED_LOG_INTERVAL);
+        if !due {
+            return None;
+        }
+        dropped.1 = Some(Instant::now());
+        Some(std::mem::take(&mut dropped.0))
+    }
+
     fn cached(&self, id: BlobId) -> Option<Arc<Prepared>> {
         self.cache
             .lock_or_recover()
@@ -126,6 +147,13 @@ pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
             let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else {
                 state.stats.blob_request_dropped();
                 debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one");
+                if let Some(dropped) = state.blobs.note_dropped() {
+                    warn!(
+                        dropped,
+                        limit = MAX_ANSWERING,
+                        "Dropped blob requests: answering too many at once (requesters ask again or elsewhere)"
+                    );
+                }
                 return;
             };
             let state = state.clone();
-- 
cgit v1.3.1