fix(ecstore): handle stalled recovery reads and listings (#3790)

* fix(ecstore): handle stalled recovery reads and listings

* fix(rio): start HTTP stall timeout on read

* fix(ecstore): handle stalled reads and partial lists

* fix(ecstore): retire stalled shards and list errors

* fix(ecstore): preserve list merge lookahead entries

* fix(ecstore): bound zero-copy shard reads

* fix(ecstore): hedge stalled shard reads

* fix(ecstore): retire abandoned shard reads

* fix(ecstore): include part identity in metadata quorum

* fix(ecstore): validate heal shard sources

* fix(ecstore): verify reconstructed read shards

* chore(ecstore): log slow object read stages

* fix(heal): throttle auto heal during recovery

* fix(scanner): yield to foreground reads

* fix(scanner): track streaming object reads

* fix(ecstore): avoid false read heal fanout

* fix(ecstore): verify codec streaming reconstruction sources

* fix(ecstore): preserve quorum progress on slow shards

* fix(storage): restore read timeout facade

* fix(ecstore): retain fallback readers after quorum

* chore: allow decode helper argument lists

---------

Co-authored-by: overtrue <anzhengchao@gmail.com>

crates/rio/src/http_reader.rs

@@ -496,7 +496,7 @@ impl HttpReader {
             headers,
             track_internode_metrics,
             internode_operation,
-            stall_timer: stall_timeout.map(|timeout| Box::pin(time::sleep(timeout))),
+            stall_timer: None,
             stall_timeout,
         })
     }
@@ -522,19 +522,15 @@ impl AsyncRead for HttpReader {
                 if bytes_read > 0 {
                     record_internode_recv_bytes(*this.track_internode_metrics, *this.internode_operation, bytes_read);
                 }
-                if bytes_read > 0 {
-                    if let Some(stall_timeout) = *this.stall_timeout {
-                        *this.stall_timer = Some(Box::pin(time::sleep(stall_timeout)));
-                    }
-                } else {
-                    *this.stall_timer = None;
-                }
+                *this.stall_timer = None;
                 Poll::Ready(Ok(()))
             }
             Poll::Pending => {
-                if let Some(timer) = this.stall_timer.as_mut()
-                    && timer.as_mut().poll(cx).is_ready()
-                {
+                let Some(stall_timeout) = *this.stall_timeout else {
+                    return Poll::Pending;
+                };
+                let timer = this.stall_timer.get_or_insert_with(|| Box::pin(time::sleep(stall_timeout)));
+                if timer.as_mut().poll(cx).is_ready() {
                     record_internode_error(*this.track_internode_metrics, *this.internode_operation);
                     Poll::Ready(Err(Error::new(
                         io::ErrorKind::TimedOut,
@@ -951,7 +947,7 @@ mod tests {
     use tokio::{
         io::{AsyncReadExt, AsyncWriteExt},
         net::TcpListener,
-        sync::Mutex,
+        sync::{Mutex, Notify},
     };
 
     #[derive(Clone, Default)]
@@ -960,6 +956,7 @@ mod tests {
         get_count: Arc<AtomicUsize>,
         put_count: Arc<AtomicUsize>,
         put_bodies: Arc<Mutex<Vec<Vec<u8>>>>,
+        delayed_body: Arc<Notify>,
     }
 
     async fn get_stream(State(state): State<TestState>) -> impl IntoResponse {
@@ -973,6 +970,16 @@ mod tests {
         (StatusCode::OK, Body::from_stream(body_stream))
     }
 
+    async fn get_delayed_first_chunk(State(state): State<TestState>) -> impl IntoResponse {
+        state.get_count.fetch_add(1, Ordering::SeqCst);
+        let delayed_body = state.delayed_body;
+        let body_stream = stream::once(async move {
+            delayed_body.notified().await;
+            Ok::<Bytes, io::Error>(Bytes::from_static(b"hello"))
+        });
+        (StatusCode::OK, Body::from_stream(body_stream))
+    }
+
     async fn reject_head(State(state): State<TestState>) -> impl IntoResponse {
         state.head_count.fetch_add(1, Ordering::SeqCst);
         StatusCode::METHOD_NOT_ALLOWED
@@ -995,6 +1002,7 @@ mod tests {
         let app = Router::new()
             .route("/stream", get(get_stream).head(reject_head).put(accept_put))
             .route("/stall", get(get_stalling_stream))
+            .route("/delayed-first", get(get_delayed_first_chunk))
             .with_state(state);
 
         let handle = tokio::spawn(async move {
@@ -1073,6 +1081,41 @@ mod tests {
         handle.abort();
     }
 
+    #[tokio::test]
+    async fn http_reader_stall_timeout_starts_when_read_is_polled() {
+        let state = TestState::default();
+        let Some((base_url, handle)) = start_test_server(state.clone()).await else {
+            return;
+        };
+        let url = base_url.replace("/stream", "/delayed-first");
+
+        let mut reader =
+            HttpReader::new_with_stall_timeout(url, Method::GET, HeaderMap::new(), None, Some(Duration::from_millis(30)))
+                .await
+                .expect("reader should be created before the body is ready");
+
+        time::sleep(Duration::from_millis(60)).await;
+
+        let delayed_body = state.delayed_body.clone();
+        let read_result = tokio::time::timeout(Duration::from_secs(1), async move {
+            let mut first = [0u8; 5];
+            let read = tokio::spawn(async move {
+                reader.read_exact(&mut first).await?;
+                Ok::<_, io::Error>(first)
+            });
+            time::sleep(Duration::from_millis(5)).await;
+            delayed_body.notify_waiters();
+            read.await.expect("read task should not panic")
+        })
+        .await
+        .expect("delayed body should arrive before the active stall timeout")
+        .expect("reader should not time out before its first active poll");
+
+        assert_eq!(&read_result, b"hello");
+
+        handle.abort();
+    }
+
     #[tokio::test]
     async fn http_writer_does_not_send_empty_preflight_put() {
         let state = TestState::default();