volume: don't let an unreachable tier backend hold up volume server startup (#12312)

* s3 tier backend: bound connect and response-header waits

The tier client used the SDK default transport: a 30s dial timeout and no
limit on waiting for response headers. A blackholed endpoint cost about
two minutes per request with the SDK's retries, and an endpoint that
accepts connections but never answers held the request forever.

Dial with a 10s timeout and give up on a response whose headers do not
arrive within a minute.

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

* volume: stop the disk scan from waiting on a stalled tier backend

A volume server registers with the master only after every volume on its
disks has loaded, and loading a tiered volume reads its superblock from the
remote object. With the S3 endpoint unreachable, each read waited out the
client's timeouts and retries on one of the loader workers, so start-up grew
with the number of tiered volumes and the local volumes were not served
until it finished.

A tiered volume already loads without its superblock when the read fails.
Give that read 10s, and once it stalls, let the backend's other volumes skip
it for a minute instead of each waiting out the timeout. The volume reads
from the backend as before once the endpoint answers again.

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

* seaweed-volume: bound connect and read waits of the S3 tier client

Mirror the Go tier client: a 10s connect timeout instead of the SDK's
3.1s default, and a minute to receive a response, so an endpoint that
accepts connections but never answers cannot hold a read forever.

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

* seaweed-volume: stop volume loading from waiting on a stalled tier backend

Mirror the Go loader: the superblock read of a tiered volume gets 10s, and
once it stalls the backend's other volumes skip it for a minute. The
volume loads with its .vif version as before when the read fails, and
reads from the backend once the endpoint answers again.

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

* seaweed-volume: keep the .vif replication when a remote superblock read fails

Go applies the .vif replication to a tiered volume whether or not its
superblock read succeeded, so a volume loaded while its backend is down
still reports its replica placement. The Rust loader applied it only on
success and reported the default placement instead.

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

* volume: skip the .idx head repair while the tier backend is unreachable

The repair reads needle headers from the remote .dat, so after the
superblock read failed it would wait on the same backend again and hold
up the load. It runs on the next load that reaches the backend.

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

* seaweed-volume: skip the .idx head repair while the tier backend is unreachable

Mirrors the Go load: after the remote superblock read failed, the
repair's remote header reads would wait on the same backend again.

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

* s3 tier backend: give up a ranged read whose body stops arriving

The response-header timeout ends once headers arrive, so an endpoint
that answers and then sends no body held ReadAt, and a superblock
read's worker, forever. Cancel the request once the body makes no
progress for s3ReadTimeout.

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

* seaweed-volume: test that a tier read whose body stalls gives up

The SDK's stalled-stream protection already fails it, matching the Go
read's no-progress cancel.

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

* volume(rust): keep the S3 read timeout off upload_part

The SDK's read timeout runs from the start of the request, so a 64 MiB
part on a slow link would time out while its body is still being sent.
Go's ResponseHeaderTimeout only starts after the body is written.
upload_part now keeps the connect timeout and drops the read timeout;
stalled-stream protection still fails an upload that stops moving.

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

* volume(rust): bound the upload_part response wait from when the part is sent

Without a read timeout, an upload_part whose body was accepted but never
answered waited forever. The part body now goes to the connection in 64 KiB
chunks, and once the last chunk is taken the response must arrive within
S3_READ_TIMEOUT, like Go's ResponseHeaderTimeout. A retry rebuilds the body
and restarts that clock. The body keeps its in-memory contents, so signing
and the checksum header are unchanged.

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

* volume(rust): keep a retried part's response clock from an earlier attempt

The send state was a single flag, and a retry body could be rebuilt and
fully sent before the waiter saw the flag drop, so the waiter kept the
earlier attempt's deadline. Each body now carries its attempt number, only
the latest attempt marks the part sent, and the deadline holds only while
no later attempt has started.

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

seaweed-volume/src/remote_storage/s3_tier.rs

@@ -5,21 +5,153 @@
 
 use std::collections::HashMap;
 use std::future::Future;
+use std::pin::Pin;
 use std::sync::{Arc, OnceLock, RwLock};
+use std::task::{Context, Poll};
+use std::time::Duration;
 
 use aws_sdk_s3::Client;
 use aws_sdk_s3::config::http::HttpResponse;
+use aws_sdk_s3::config::timeout::TimeoutConfig;
 use aws_sdk_s3::config::{BehaviorVersion, Credentials, Region};
 use aws_sdk_s3::error::{DisplayErrorContext, SdkError};
 use aws_sdk_s3::operation::get_object::GetObjectError;
 use aws_sdk_s3::operation::head_object::HeadObjectError;
+use aws_sdk_s3::operation::upload_part::UploadPartOutput;
+use aws_sdk_s3::operation::upload_part::builders::UploadPartFluentBuilder;
+use aws_sdk_s3::primitives::{ByteStream, SdkBody};
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+use bytes::Bytes;
 use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
-use tokio::sync::Semaphore;
+use tokio::sync::{Semaphore, watch};
 
 /// Concurrency limit for multipart upload/download (matches Go's s3manager).
 const CONCURRENCY: usize = 5;
 
+/// A tier endpoint that drops packets or accepts connections without answering
+/// must not hold a read, or a volume server's startup, for minutes per request.
+/// Matches Go's s3ConnectTimeout and s3ReadTimeout. The SDK's stalled-stream
+/// protection already fails a body that stops arriving.
+const S3_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+const S3_READ_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// The SDK's read timeout runs from the start of the request, so on
+/// upload_part it would also cap sending the part body. send_part bounds the
+/// response wait instead, from when the body is written, like Go's
+/// ResponseHeaderTimeout.
+fn upload_part_config() -> aws_sdk_s3::config::Builder {
+    aws_sdk_s3::config::Builder::default().timeout_config(
+        TimeoutConfig::builder()
+            .connect_timeout(S3_CONNECT_TIMEOUT)
+            .disable_read_timeout()
+            .build(),
+    )
+}
+
+/// A part goes to the connection in chunks this size, so its last chunk is
+/// taken only once nearly all of the part has been written.
+const UPLOAD_CHUNK_SIZE: usize = 64 * 1024;
+
+/// The latest attempt to send a part, and whether its body has been sent.
+type PartSent = Arc<watch::Sender<(u64, bool)>>;
+
+struct PartBody {
+    rest: Bytes,
+    attempt: u64,
+    sent: PartSent,
+}
+
+impl http_body::Body for PartBody {
+    type Data = Bytes;
+    type Error = std::convert::Infallible;
+
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
+        if self.rest.is_empty() {
+            return Poll::Ready(None);
+        }
+        let n = self.rest.len().min(UPLOAD_CHUNK_SIZE);
+        let chunk = self.rest.split_to(n);
+        if self.rest.is_empty() {
+            let attempt = self.attempt;
+            self.sent.send_if_modified(|state| {
+                let latest = state.0 == attempt;
+                state.1 |= latest;
+                latest
+            });
+        }
+        Poll::Ready(Some(Ok(http_body::Frame::data(chunk))))
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.rest.is_empty()
+    }
+
+    fn size_hint(&self) -> http_body::SizeHint {
+        http_body::SizeHint::with_exact(self.rest.len() as u64)
+    }
+}
+
+/// Sends one multipart part, giving up once `response_timeout` passes after
+/// the whole part was handed to the connection without a response. A retry
+/// rebuilds the body and restarts that clock. The body keeps its in-memory
+/// contents, so signing and the checksum header are unchanged.
+async fn send_part(
+    request: UploadPartFluentBuilder,
+    data: Bytes,
+    response_timeout: Duration,
+) -> Result<UploadPartOutput, String> {
+    let sent: PartSent = Arc::new(watch::channel((0, false)).0);
+    let body = {
+        let sent = sent.clone();
+        SdkBody::from(data.clone()).map_preserve_contents(move |_| {
+            let mut attempt = 0;
+            sent.send_modify(|state| {
+                *state = (state.0 + 1, data.is_empty());
+                attempt = state.0;
+            });
+            SdkBody::from_body_1_x(PartBody {
+                rest: data.clone(),
+                attempt,
+                sent: sent.clone(),
+            })
+        })
+    };
+    tokio::select! {
+        resp = request
+            .body(ByteStream::new(body))
+            .customize()
+            .config_override(upload_part_config())
+            .send() => resp.map_err(|e| DisplayErrorContext(&e).to_string()),
+        () = wait_unanswered(sent, response_timeout) => Err(format!(
+            "no response within {response_timeout:?} after sending the part"
+        )),
+    }
+}
+
+/// Completes once an attempt's body has been sent and `response_timeout`
+/// passes before a later attempt starts.
+async fn wait_unanswered(sent: PartSent, response_timeout: Duration) {
+    let mut state = sent.subscribe();
+    loop {
+        let attempt = state
+            .wait_for(|&(_, done)| done)
+            .await
+            .map_or(0, |state| state.0);
+        if tokio::time::timeout(
+            response_timeout,
+            state.wait_for(|&(latest, _)| latest != attempt),
+        )
+        .await
+        .is_err()
+        {
+            return;
+        }
+    }
+}
+
 /// A tier transfer failure. The variant is what callers match on; the
 /// message is the operator-facing text.
 #[derive(Debug, thiserror::Error)]
@@ -107,7 +239,13 @@ impl S3TierBackend {
             .behavior_version(BehaviorVersion::latest())
             .region(Region::new(region.to_string()))
             .credentials_provider(credentials)
-            .force_path_style(config.force_path_style);
+            .force_path_style(config.force_path_style)
+            .timeout_config(
+                TimeoutConfig::builder()
+                    .connect_timeout(S3_CONNECT_TIMEOUT)
+                    .read_timeout(S3_READ_TIMEOUT)
+                    .build(),
+            );
 
         if !config.endpoint.is_empty() {
             s3_config = s3_config.endpoint_url(&config.endpoint);
@@ -231,23 +369,23 @@ impl S3TierBackend {
                     TierError::Io(format!("failed to read file at offset {}: {}", off, e))
                 })?;
 
-                let upload_part_resp = client
-                    .upload_part()
-                    .bucket(&bucket)
-                    .key(&key)
-                    .upload_id(&upload_id)
-                    .part_number(pn)
-                    .body(buf.into())
-                    .send()
-                    .await
-                    .map_err(|e| {
-                        TierError::Io(format!(
-                            "failed to upload part {} at offset {}: {}",
-                            pn,
-                            off,
-                            DisplayErrorContext(&e)
-                        ))
-                    })?;
+                let upload_part_resp = send_part(
+                    client
+                        .upload_part()
+                        .bucket(&bucket)
+                        .key(&key)
+                        .upload_id(&upload_id)
+                        .part_number(pn),
+                    buf.into(),
+                    S3_READ_TIMEOUT,
+                )
+                .await
+                .map_err(|e| {
+                    TierError::Io(format!(
+                        "failed to upload part {} at offset {}: {}",
+                        pn, off, e
+                    ))
+                })?;
 
                 let e_tag = upload_part_resp.e_tag().unwrap_or_default().to_string();
 
@@ -560,6 +698,25 @@ impl S3TierBackend {
         })
     }
 
+    /// `read_range_blocking` that gives up once `timeout` passes, returning
+    /// `Ok(None)`.
+    pub fn read_range_blocking_within(
+        self: &Arc<Self>,
+        key: &str,
+        offset: u64,
+        size: usize,
+        timeout: Duration,
+    ) -> Result<Option<Vec<u8>>, TierError> {
+        let backend = Arc::clone(self);
+        let key = key.to_string();
+        block_on_tier_future(async move {
+            match tokio::time::timeout(timeout, backend.read_range(&key, offset, size)).await {
+                Ok(read) => read.map(Some),
+                Err(_) => Ok(None),
+            }
+        })
+    }
+
     pub fn read_range_blocking(
         &self,
         key: &str,
@@ -725,6 +882,13 @@ fn describe_join_error(join_error: tokio::task::JoinError) -> TierError {
 mod tests {
     use super::*;
     use crate::remote_storage::s3::tests::{CannedResponse, NO_SUCH_KEY};
+    use aws_sdk_s3::config::http::HttpRequest;
+    use aws_sdk_s3::config::{HttpClient, RuntimeComponents};
+    use aws_smithy_runtime_api::client::http::{
+        HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector,
+    };
+    use aws_smithy_runtime_api::http::StatusCode;
+    use http_body::Body as _;
     use std::collections::HashSet;
     use tokio::runtime::Handle;
 
@@ -897,6 +1061,235 @@ mod tests {
         assert!(matches!(err, TierError::Io(_)), "{err:?}");
     }
 
+    #[test]
+    fn backend_bounds_connect_and_read_waits() {
+        let (_listener, backend) = silent_backend();
+        let timeouts = backend.client.config().timeout_config().unwrap();
+        assert_eq!(timeouts.connect_timeout(), Some(S3_CONNECT_TIMEOUT));
+        assert_eq!(timeouts.read_timeout(), Some(S3_READ_TIMEOUT));
+    }
+
+    #[derive(Debug)]
+    struct CaptureTimeouts(Arc<std::sync::Mutex<Option<TimeoutConfig>>>);
+
+    impl aws_sdk_s3::config::Intercept for CaptureTimeouts {
+        fn name(&self) -> &'static str {
+            "CaptureTimeouts"
+        }
+
+        fn read_before_transmit(
+            &self,
+            _context: &aws_sdk_s3::config::interceptors::BeforeTransmitInterceptorContextRef<'_>,
+            _runtime_components: &aws_sdk_s3::config::RuntimeComponents,
+            cfg: &mut aws_sdk_s3::config::ConfigBag,
+        ) -> Result<(), aws_sdk_s3::error::BoxError> {
+            *self.0.lock().unwrap() = cfg.load::<TimeoutConfig>().cloned();
+            Err("captured".into())
+        }
+    }
+
+    #[tokio::test]
+    async fn upload_part_keeps_the_connect_timeout_without_the_read_timeout() {
+        let (_listener, backend) = silent_backend();
+        let seen = Arc::new(std::sync::Mutex::new(None));
+        let _ = backend
+            .client
+            .upload_part()
+            .bucket("bucket")
+            .key("key")
+            .upload_id("upload")
+            .part_number(1)
+            .body(vec![0u8; 8].into())
+            .customize()
+            .config_override(upload_part_config().interceptor(CaptureTimeouts(seen.clone())))
+            .send()
+            .await;
+        let timeouts = seen
+            .lock()
+            .unwrap()
+            .clone()
+            .expect("request never reached transmit");
+        assert_eq!(timeouts.connect_timeout(), Some(S3_CONNECT_TIMEOUT));
+        assert_eq!(timeouts.read_timeout(), None);
+    }
+
+    fn upload_part_request(backend: &S3TierBackend) -> UploadPartFluentBuilder {
+        backend
+            .client
+            .upload_part()
+            .bucket("bucket")
+            .key("key")
+            .upload_id("upload")
+            .part_number(1)
+    }
+
+    #[tokio::test]
+    async fn send_part_gives_up_when_no_response_follows_the_body() {
+        let (_listener, backend) = silent_backend();
+        let sent = tokio::time::timeout(
+            Duration::from_secs(30),
+            send_part(
+                upload_part_request(&backend),
+                Bytes::from_static(&[0u8; 8]),
+                Duration::from_millis(200),
+            ),
+        )
+        .await
+        .expect("a part with no response did not give up");
+        let err = sent.unwrap_err();
+        assert!(err.starts_with("no response within 200ms"), "{err}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_retry_restarts_the_response_clock() {
+        let sent: PartSent = Arc::new(watch::channel((1, true)).0);
+        let mut unanswered =
+            std::pin::pin!(wait_unanswered(sent.clone(), Duration::from_millis(200)));
+        let wait = |ms| tokio::time::sleep(Duration::from_millis(ms));
+        tokio::select! {
+            () = unanswered.as_mut() => panic!("gave up early"),
+            () = wait(150) => {}
+        }
+        // The retry starts and finishes sending before the waiter runs again.
+        sent.send_replace((2, false));
+        sent.send_replace((2, true));
+        tokio::select! {
+            () = unanswered.as_mut() => panic!("kept the first attempt's clock"),
+            () = wait(150) => {}
+        }
+        tokio::select! {
+            () = unanswered => {}
+            () = wait(100) => panic!("the retry never timed out"),
+        }
+    }
+
+    /// Takes the request body one chunk per `pause`, then answers at once.
+    #[derive(Debug, Clone)]
+    struct SlowReader {
+        pause: Duration,
+    }
+
+    impl HttpConnector for SlowReader {
+        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+            let pause = self.pause;
+            HttpConnectorFuture::new(async move {
+                let mut body = std::pin::pin!(request.into_body());
+                loop {
+                    tokio::time::sleep(pause).await;
+                    let frame = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await;
+                    if frame.is_none() || body.is_end_stream() {
+                        break;
+                    }
+                }
+                let mut response =
+                    HttpResponse::new(StatusCode::try_from(200).unwrap(), SdkBody::empty());
+                response.headers_mut().insert("ETag", "\"etag\"");
+                Ok(response)
+            })
+        }
+    }
+
+    impl HttpClient for SlowReader {
+        fn http_connector(
+            &self,
+            _settings: &HttpConnectorSettings,
+            _components: &RuntimeComponents,
+        ) -> SharedHttpConnector {
+            SharedHttpConnector::new(self.clone())
+        }
+    }
+
+    #[tokio::test]
+    async fn send_part_does_not_cap_a_slow_body() {
+        let config = aws_sdk_s3::Config::builder()
+            .behavior_version(BehaviorVersion::latest())
+            .region(Region::new("us-east-1"))
+            .credentials_provider(Credentials::new("AKIATEST", "secret", None, None, "test"))
+            .endpoint_url("http://127.0.0.1:1")
+            .force_path_style(true)
+            .http_client(SlowReader {
+                pause: Duration::from_millis(300),
+            })
+            .build();
+        let backend = S3TierBackend {
+            client: Client::from_conf(config),
+            bucket: "bucket".to_string(),
+            storage_class: "STANDARD".to_string(),
+        };
+        let output = send_part(
+            upload_part_request(&backend),
+            Bytes::from(vec![0u8; 4 * UPLOAD_CHUNK_SIZE]),
+            Duration::from_millis(200),
+        )
+        .await
+        .expect("a part still being sent was cut short");
+        assert_eq!(output.e_tag(), Some("\"etag\""));
+    }
+
+    #[test]
+    fn read_range_blocking_within_gives_up_on_a_silent_endpoint() {
+        let (_listener, backend) = silent_backend();
+        let start = std::time::Instant::now();
+        let got = backend
+            .read_range_blocking_within("key", 0, 8, Duration::from_millis(200))
+            .unwrap();
+        assert!(got.is_none());
+        assert!(start.elapsed() < Duration::from_secs(5));
+    }
+
+    #[tokio::test]
+    async fn read_range_gives_up_when_the_body_stalls() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let endpoint = format!("http://{}", listener.local_addr().unwrap());
+        std::thread::spawn(move || {
+            let mut held = Vec::new();
+            for mut conn in listener.incoming().flatten() {
+                let mut request = Vec::new();
+                let mut byte = [0u8; 1];
+                while !request.ends_with(b"\r\n\r\n") {
+                    match std::io::Read::read(&mut conn, &mut byte) {
+                        Ok(1) => request.push(byte[0]),
+                        _ => break,
+                    }
+                }
+                let _ = std::io::Write::write_all(
+                    &mut conn,
+                    b"HTTP/1.1 206 Partial Content\r\nContent-Length: 8\r\nContent-Range: bytes 0-7/1024\r\n\r\n",
+                );
+                held.push(conn);
+            }
+        });
+        let backend = S3TierBackend::new(&S3TierConfig {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            region: "us-east-1".to_string(),
+            bucket: "bucket".to_string(),
+            endpoint,
+            storage_class: String::new(),
+            force_path_style: true,
+        });
+
+        let read = tokio::time::timeout(Duration::from_secs(30), backend.read_range("key", 0, 8))
+            .await
+            .expect("a read whose body stalls did not give up");
+        let err = read.unwrap_err();
+        assert!(matches!(err, TierError::Io(_)), "{err:?}");
+    }
+
+    fn silent_backend() -> (std::net::TcpListener, Arc<S3TierBackend>) {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let backend = S3TierBackend::new(&S3TierConfig {
+            access_key: "access".to_string(),
+            secret_key: "secret".to_string(),
+            region: "us-east-1".to_string(),
+            bucket: "bucket".to_string(),
+            endpoint: format!("http://{}", listener.local_addr().unwrap()),
+            storage_class: String::new(),
+            force_path_style: true,
+        });
+        (listener, Arc::new(backend))
+    }
+
     #[test]
     fn read_range_blocking_no_such_key_is_not_found() {
         let err = backend_answering(404, NO_SUCH_KEY)

seaweed-volume/src/storage/volume.rs

@@ -9,14 +9,14 @@
 //! Matches Go's storage/volume.go, volume_loading.go, volume_read.go,
 //! volume_write.go, volume_super_block.go.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::fs::{self, File, OpenOptions};
 use std::io::{self, Read, Seek, SeekFrom, Write};
 use std::ops::ControlFlow;
 use std::path::Path;
 use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64, Ordering};
-use std::sync::{Arc, Condvar, Mutex, RwLock};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::sync::{Arc, Condvar, LazyLock, Mutex, RwLock};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use tracing::{debug, error, info, warn};
 
@@ -1198,6 +1198,33 @@ impl RemoteDatFile {
             .backend
             .read_range_blocking(&self.key, offset, buf.len())
             .map_err(io::Error::other)?;
+        Self::fill(buf, offset, data)
+    }
+
+    /// `read_exact_at` that fails with `TimedOut` once `timeout` passes.
+    pub(crate) fn read_exact_at_within(
+        &self,
+        buf: &mut [u8],
+        offset: u64,
+        timeout: Duration,
+    ) -> io::Result<()> {
+        match self
+            .backend
+            .read_range_blocking_within(&self.key, offset, buf.len(), timeout)
+            .map_err(io::Error::other)?
+        {
+            Some(data) => Self::fill(buf, offset, data),
+            None => Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                format!(
+                    "remote read at offset {} got no answer within {:?}",
+                    offset, timeout
+                ),
+            )),
+        }
+    }
+
+    fn fill(buf: &mut [u8], offset: u64, data: Vec<u8>) -> io::Result<()> {
         if data.len() != buf.len() {
             return Err(io::Error::new(
                 io::ErrorKind::UnexpectedEof,
@@ -1214,6 +1241,71 @@ impl RemoteDatFile {
     }
 }
 
+/// A tiered volume loads without its superblock when the backend cannot serve
+/// it; once a read stalls, the backend's other volumes skip it for a while.
+/// Matches Go's remoteSuperBlockTimeout and remoteSuperBlockStallPeriod.
+const REMOTE_SUPER_BLOCK_TIMEOUT: Duration = Duration::from_secs(10);
+const REMOTE_SUPER_BLOCK_STALL_PERIOD: Duration = Duration::from_secs(60);
+static REMOTE_SUPER_BLOCK_STALLS: LazyLock<Mutex<HashMap<String, Instant>>> =
+    LazyLock::new(Default::default);
+
+fn read_super_block_bytes(
+    mut read: impl FnMut(&mut [u8], u64) -> Result<(), VolumeError>,
+) -> Result<Vec<u8>, VolumeError> {
+    let mut header = [0u8; SUPER_BLOCK_SIZE];
+    read(&mut header, 0)?;
+
+    let extra_size = u16::from_be_bytes([header[6], header[7]]);
+    let mut full_buf = vec![0u8; SUPER_BLOCK_SIZE + extra_size as usize];
+    full_buf[..SUPER_BLOCK_SIZE].copy_from_slice(&header);
+    if extra_size > 0 {
+        read(&mut full_buf[SUPER_BLOCK_SIZE..], SUPER_BLOCK_SIZE as u64)?;
+    }
+    Ok(full_buf)
+}
+
+fn read_remote_super_block(
+    storage_name: &str,
+    remote: &RemoteDatFile,
+    timeout: Duration,
+    stall_period: Duration,
+) -> Result<Vec<u8>, VolumeError> {
+    if let Some(until) = REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().get(storage_name)
+        && Instant::now() < *until
+    {
+        return Err(VolumeError::Io(io::Error::other(format!(
+            "backend {} stalled on a super block read, skipped",
+            storage_name
+        ))));
+    }
+    let deadline = Instant::now() + timeout;
+    let result = read_super_block_bytes(|buf, offset| {
+        remote
+            .read_exact_at_within(
+                buf,
+                offset,
+                deadline.saturating_duration_since(Instant::now()),
+            )
+            .map_err(VolumeError::from)
+    });
+    match &result {
+        Ok(_) => {
+            REMOTE_SUPER_BLOCK_STALLS
+                .lock()
+                .unwrap()
+                .remove(storage_name);
+        }
+        Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::TimedOut => {
+            REMOTE_SUPER_BLOCK_STALLS
+                .lock()
+                .unwrap()
+                .insert(storage_name.to_string(), Instant::now() + stall_period);
+        }
+        Err(_) => {}
+    }
+    result
+}
+
 // ============================================================================
 // Volume
 // ============================================================================
@@ -1684,6 +1776,7 @@ impl Volume {
             )));
         }
 
+        let mut remote_super_block_failed = false;
         if already_has_super_block {
             match self.read_super_block(io) {
                 Ok(()) => {
@@ -1700,6 +1793,12 @@ impl Volume {
                         error = %e,
                         "failed to read remote super block during load"
                     );
+                    remote_super_block_failed = true;
+                    let replication = self.volume_info.read().unwrap().replication.clone();
+                    if !replication.is_empty() {
+                        self.super_block.write().unwrap().replica_placement =
+                            ReplicaPlacement::from_string(&replication)?;
+                    }
                 }
                 Err(e) => return Err(e),
             }
@@ -1748,17 +1847,24 @@ impl Volume {
             // at the front of .idx. Best effort: a volume that cannot be
             // repaired is still servable for everything the surviving rows
             // index.
-            match self.repair_idx_head_tombstones(io) {
-                Ok(0) => {}
-                Ok(restored) => info!(
+            if remote_super_block_failed {
+                debug!(
                     volume_id = self.id.0,
-                    restored, "recovered overwritten .idx rows from .dat"
-                ),
-                Err(e) => warn!(
-                    volume_id = self.id.0,
-                    error = %e,
-                    "recover overwritten .idx rows"
-                ),
+                    "remote backend unavailable, not recovering overwritten .idx rows on this load"
+                );
+            } else {
+                match self.repair_idx_head_tombstones(io) {
+                    Ok(0) => {}
+                    Ok(restored) => info!(
+                        volume_id = self.id.0,
+                        restored, "recovered overwritten .idx rows from .dat"
+                    ),
+                    Err(e) => warn!(
+                        volume_id = self.id.0,
+                        error = %e,
+                        "recover overwritten .idx rows"
+                    ),
+                }
             }
 
             // The permission bits still allow writing on a read-only mount;
@@ -2220,21 +2326,15 @@ impl Volume {
     // ---- SuperBlock I/O ----
 
     fn read_super_block(&self, io: &VolumeIo) -> Result<(), VolumeError> {
-        let mut header = [0u8; SUPER_BLOCK_SIZE];
-        self.read_exact_at_backend(io, &mut header, 0)?;
-
-        let extra_size = u16::from_be_bytes([header[6], header[7]]);
-        let total_size = SUPER_BLOCK_SIZE + extra_size as usize;
-
-        let mut full_buf = vec![0u8; total_size];
-        full_buf[..SUPER_BLOCK_SIZE].copy_from_slice(&header);
-        if extra_size > 0 {
-            self.read_exact_at_backend(
-                io,
-                &mut full_buf[SUPER_BLOCK_SIZE..],
-                SUPER_BLOCK_SIZE as u64,
-            )?;
-        }
+        let full_buf = match io.remote_dat_file.as_ref() {
+            Some(remote) if self.has_remote_file() => read_remote_super_block(
+                &self.remote_storage_name_key().0,
+                remote,
+                REMOTE_SUPER_BLOCK_TIMEOUT,
+                REMOTE_SUPER_BLOCK_STALL_PERIOD,
+            )?,
+            _ => read_super_block_bytes(|buf, offset| self.read_exact_at_backend(io, buf, offset))?,
+        };
 
         *self.super_block.write().unwrap() = SuperBlock::from_bytes(&full_buf)?;
 
@@ -11516,6 +11616,254 @@ mod tests {
         assert!(result.is_err(), "unopenable .dat must fail load");
     }
 
+    fn silent_tier_backend() -> (
+        std::net::TcpListener,
+        Arc<crate::remote_storage::s3_tier::S3TierBackend>,
+    ) {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let backend = crate::remote_storage::s3_tier::S3TierBackend::new(
+            &crate::remote_storage::s3_tier::S3TierConfig {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                region: "us-east-1".to_string(),
+                bucket: "bucket-a".to_string(),
+                endpoint: format!("http://{}", listener.local_addr().unwrap()),
+                storage_class: "STANDARD".to_string(),
+                force_path_style: true,
+            },
+        );
+        (listener, Arc::new(backend))
+    }
+
+    #[test]
+    fn remote_super_block_read_gives_up_on_a_silent_backend_and_skips_it() {
+        let name = "s3.super_block_stall_test";
+        let (_listener, backend) = silent_tier_backend();
+        let silent = RemoteDatFile {
+            backend,
+            key: "remote-key".to_string(),
+            file_size: 1024,
+            modified_time: 0,
+        };
+
+        let start = Instant::now();
+        let err = read_remote_super_block(
+            name,
+            &silent,
+            Duration::from_millis(200),
+            Duration::from_secs(3600),
+        )
+        .unwrap_err();
+        assert!(
+            matches!(&err, VolumeError::Io(e) if e.kind() == io::ErrorKind::TimedOut),
+            "{err:?}"
+        );
+        assert!(start.elapsed() < Duration::from_secs(5));
+
+        let start = Instant::now();
+        assert!(
+            read_remote_super_block(
+                name,
+                &silent,
+                Duration::from_secs(30),
+                Duration::from_secs(3600)
+            )
+            .is_err()
+        );
+        assert!(
+            start.elapsed() < Duration::from_secs(1),
+            "a stalled backend must not be asked again within the stall period"
+        );
+
+        let sb = SuperBlock {
+            version: Version::current(),
+            ..SuperBlock::default()
+        };
+        let (endpoint, shutdown_tx) = spawn_fake_s3_server(sb.to_bytes());
+        let answering = RemoteDatFile {
+            backend: Arc::new(crate::remote_storage::s3_tier::S3TierBackend::new(
+                &crate::remote_storage::s3_tier::S3TierConfig {
+                    access_key: "access".to_string(),
+                    secret_key: "secret".to_string(),
+                    region: "us-east-1".to_string(),
+                    bucket: "bucket-a".to_string(),
+                    endpoint,
+                    storage_class: "STANDARD".to_string(),
+                    force_path_style: true,
+                },
+            )),
+            key: "remote-key".to_string(),
+            file_size: SUPER_BLOCK_SIZE as u64,
+            modified_time: 0,
+        };
+        REMOTE_SUPER_BLOCK_STALLS
+            .lock()
+            .unwrap()
+            .insert(name.to_string(), Instant::now());
+        let bytes = read_remote_super_block(
+            name,
+            &answering,
+            Duration::from_secs(10),
+            Duration::from_secs(3600),
+        )
+        .unwrap();
+        assert_eq!(
+            SuperBlock::from_bytes(&bytes).unwrap().version,
+            Version::current()
+        );
+        assert!(
+            !REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().contains_key(name),
+            "a successful read clears the stall"
+        );
+        let _ = shutdown_tx.send(());
+    }
+
+    #[test]
+    fn remote_volume_loads_without_waiting_on_a_stalled_backend() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let dat_len = {
+            let v = make_test_volume(dir);
+            v.dat_file_size().unwrap()
+        };
+        std::fs::remove_file(format!("{}/1.dat", dir)).unwrap();
+
+        let (_listener, backend) = silent_tier_backend();
+        let name = "s3.load_stall_test";
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .register(name.to_string(), Arc::try_unwrap(backend).ok().unwrap());
+        REMOTE_SUPER_BLOCK_STALLS
+            .lock()
+            .unwrap()
+            .insert(name.to_string(), Instant::now() + Duration::from_secs(3600));
+
+        let vif = VifVolumeInfo {
+            files: vec![VifRemoteFile {
+                backend_type: "s3".to_string(),
+                backend_id: "load_stall_test".to_string(),
+                key: "remote-key".to_string(),
+                offset: 0,
+                file_size: dat_len,
+                modified_time: 123,
+                extension: ".dat".to_string(),
+            }],
+            version: Version::current().0 as u32,
+            replication: "001".to_string(),
+            ..VifVolumeInfo::default()
+        };
+        std::fs::write(
+            format!("{}/1.vif", dir),
+            serde_json::to_string_pretty(&vif).unwrap(),
+        )
+        .unwrap();
+
+        let start = Instant::now();
+        let v = Volume::new(
+            dir,
+            dir,
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            &VolumeSpec::default(),
+        )
+        .unwrap();
+        assert!(start.elapsed() < Duration::from_secs(5));
+        assert!(v.has_remote_file());
+        assert_eq!(v.version(), Version::current());
+        assert_eq!(
+            v.super_block.read().unwrap().replica_placement,
+            ReplicaPlacement::from_string("001").unwrap()
+        );
+
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove(name);
+        REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().remove(name);
+    }
+
+    #[test]
+    fn remote_volume_load_skips_idx_repair_on_a_stalled_backend() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let dat_len = {
+            let v = make_test_volume(dir);
+            for i in 1..=3u64 {
+                let mut n = Needle {
+                    id: NeedleId(i),
+                    cookie: Cookie(0x1234),
+                    data: format!("needle-{i}").into_bytes(),
+                    data_size: format!("needle-{i}").len() as u32,
+                    ..Needle::default()
+                };
+                v.write_needle(&mut n, true, false).unwrap();
+            }
+            v.dat_file_size().unwrap()
+        };
+        std::fs::remove_file(format!("{}/1.dat", dir)).unwrap();
+        let idx_path = format!("{}/1.idx", dir);
+        crate::storage::volume_idx_repair::tests::clobber_idx_head(&idx_path, &[4, 5]);
+        let idx_before = std::fs::read(&idx_path).unwrap();
+
+        let (_listener, backend) = silent_tier_backend();
+        let name = "s3.idx_repair_stall_test";
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .register(name.to_string(), Arc::try_unwrap(backend).ok().unwrap());
+        REMOTE_SUPER_BLOCK_STALLS
+            .lock()
+            .unwrap()
+            .insert(name.to_string(), Instant::now() + Duration::from_secs(3600));
+
+        let vif = VifVolumeInfo {
+            files: vec![VifRemoteFile {
+                backend_type: "s3".to_string(),
+                backend_id: "idx_repair_stall_test".to_string(),
+                key: "remote-key".to_string(),
+                offset: 0,
+                file_size: dat_len,
+                modified_time: 123,
+                extension: ".dat".to_string(),
+            }],
+            version: Version::current().0 as u32,
+            ..VifVolumeInfo::default()
+        };
+        std::fs::write(
+            format!("{}/1.vif", dir),
+            serde_json::to_string_pretty(&vif).unwrap(),
+        )
+        .unwrap();
+
+        let start = Instant::now();
+        let v = Volume::new(
+            dir,
+            dir,
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            &VolumeSpec::default(),
+        )
+        .unwrap();
+        assert!(
+            start.elapsed() < Duration::from_secs(5),
+            "the load waited on the stalled backend"
+        );
+        assert!(v.has_remote_file());
+        drop(v);
+        assert_eq!(
+            std::fs::read(&idx_path).unwrap(),
+            idx_before,
+            "the repair ran against an unreachable backend"
+        );
+
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove(name);
+        REMOTE_SUPER_BLOCK_STALLS.lock().unwrap().remove(name);
+    }
+
     #[test]
     fn test_remote_only_volume_load_reads_from_tier_backend() {
         let tmp = TempDir::new().unwrap();

seaweed-volume/src/storage/volume_idx_repair.rs

@@ -224,7 +224,7 @@ fn finish_walk(walk: io::Result<()>) -> Result<(), VolumeError> {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
     use crate::storage::needle::crc::CRC;
     use crate::storage::needle_map::NeedleMapKind;
@@ -267,7 +267,7 @@ mod tests {
     /// Reproduce the damage a delete on a tiered read-only volume used to do: it
     /// writes (key, offset 0, tombstone) rows over the front of .idx instead of
     /// appending them.
-    fn clobber_idx_head(idx_path: &str, keys: &[u64]) {
+    pub(crate) fn clobber_idx_head(idx_path: &str, keys: &[u64]) {
         let mut file = OpenOptions::new().write(true).open(idx_path).unwrap();
         for (i, key) in keys.iter().enumerate() {
             let mut row = Vec::new();

weed/storage/backend/s3_backend/s3_backend.go

@@ -1,6 +1,7 @@
 package s3_backend
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"math"
@@ -222,7 +223,9 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n
 
 	bytesRange := fmt.Sprintf("bytes=%d-%d", off, off+int64(len(p))-1)
 
-	getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObject(&s3.GetObjectInput{
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObjectWithContext(ctx, &s3.GetObjectInput{
 		Bucket: &s3backendStorageFile.backendStorage.bucket,
 		Key:    &s3backendStorageFile.key,
 		Range:  &bytesRange,
@@ -232,6 +235,8 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n
 		return 0, fmt.Errorf("bucket %s GetObject %s: %v", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, getObjectErr)
 	}
 	defer getObjectOutput.Body.Close()
+	stalled := time.AfterFunc(s3ReadTimeout, cancel)
+	defer stalled.Stop()
 
 	// glog.V(3).Infof("read %s %s", s3backendStorageFile.key, bytesRange)
 	// glog.V(3).Infof("content range: %s, contentLength: %d", *getObjectOutput.ContentRange, *getObjectOutput.ContentLength)
@@ -242,6 +247,7 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n
 		if len(p) == 0 {
 			break
 		}
+		stalled.Reset(s3ReadTimeout)
 		readCount, err = getObjectOutput.Body.Read(p)
 		n += readCount
 
@@ -253,6 +259,9 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n
 	if err == io.EOF && n == requested {
 		err = nil
 	}
+	if err != nil && ctx.Err() != nil {
+		err = fmt.Errorf("bucket %s read %s: no data for %v: %w", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, s3ReadTimeout, err)
+	}
 
 	return
 }

weed/storage/backend/s3_backend/s3_backend_test.go

@@ -7,6 +7,8 @@ import (
 	"testing"
 	"testing/iotest"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
@@ -23,7 +25,7 @@ type stubS3Client struct {
 	getObject func(*s3.GetObjectInput) (*s3.GetObjectOutput, error)
 }
 
-func (s *stubS3Client) GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error) {
+func (s *stubS3Client) GetObjectWithContext(_ aws.Context, input *s3.GetObjectInput, _ ...request.Option) (*s3.GetObjectOutput, error) {
 	return s.getObject(input)
 }
 

weed/storage/backend/s3_backend/s3_sessions.go

@@ -3,6 +3,7 @@ package s3_backend
 import (
 	"crypto/tls"
 	"fmt"
+	"net"
 	"net/http"
 	"sync"
 	"time"
@@ -21,8 +22,14 @@ var (
 	sessionsLock sync.Mutex
 )
 
-// httpOptions tunes the HTTP transport of the tier backend's S3 client. The
-// zero value keeps the SDK's default client.
+// A tier endpoint that drops packets or accepts connections without answering
+// must not hold a read, or a volume server's startup, for minutes per request.
+var (
+	s3ConnectTimeout = 10 * time.Second
+	s3ReadTimeout    = time.Minute
+)
+
+// httpOptions tunes the HTTP transport of the tier backend's S3 client.
 type httpOptions struct {
 	// disableHTTP2 keeps the client on HTTP/1.1. Over HTTP/2 every upload and
 	// range read to the endpoint shares one connection and its flow-control
@@ -36,12 +43,10 @@ type httpOptions struct {
 	idleConnTimeout time.Duration
 }
 
-// newTransport returns the transport for the options, or nil for the default.
 func (o httpOptions) newTransport() *http.Transport {
-	if o == (httpOptions{}) {
-		return nil
-	}
 	t := http.DefaultTransport.(*http.Transport).Clone()
+	t.DialContext = (&net.Dialer{Timeout: s3ConnectTimeout, KeepAlive: 30 * time.Second}).DialContext
+	t.ResponseHeaderTimeout = s3ReadTimeout
 	if o.disableHTTP2 {
 		t.ForceAttemptHTTP2 = false
 		t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{}
@@ -73,9 +78,7 @@ func createSession(awsAccessKeyId, awsSecretAccessKey, region, endpoint string,
 		Endpoint:                      aws.String(endpoint),
 		S3ForcePathStyle:              aws.Bool(forcePathStyle),
 		S3DisableContentMD5Validation: aws.Bool(true),
-	}
-	if t := httpOpts.newTransport(); t != nil {
-		config.HTTPClient = &http.Client{Transport: t}
+		HTTPClient:                    &http.Client{Transport: httpOpts.newTransport()},
 	}
 	if awsAccessKeyId != "" && awsSecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "")

weed/storage/backend/s3_backend/s3_sessions_test.go

@@ -1,13 +1,18 @@
 package s3_backend
 
 import (
+	"bufio"
 	"math"
+	"net"
 	"net/http"
 	"strconv"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
 func backendTransport(t *testing.T, s *S3BackendStorage) *http.Transport {
@@ -54,17 +59,103 @@ func TestS3BackendHTTPOptionsReachTheClient(t *testing.T) {
 	}
 }
 
-func TestS3BackendDefaultHTTPClientUnchanged(t *testing.T) {
+func TestS3BackendDefaultTransportHasTimeouts(t *testing.T) {
 	s, err := newS3BackendStorage(testProperties{"endpoint": "http://127.0.0.1:2"}, "", "http-defaults")
 	if err != nil {
 		t.Fatal(err)
 	}
-	if transport := backendTransport(t, s); transport != nil && transport != http.DefaultTransport {
-		t.Fatalf("no options set, yet the client got a custom transport %+v", transport)
+	transport := backendTransport(t, s)
+	if transport == nil || transport == http.DefaultTransport {
+		t.Fatalf("the client must not use the shared default transport, got %+v", transport)
 	}
-	if (httpOptions{}).newTransport() != nil {
-		t.Fatal("zero options must keep the default client")
+	if transport.ResponseHeaderTimeout != s3ReadTimeout {
+		t.Errorf("ResponseHeaderTimeout = %v, want %v", transport.ResponseHeaderTimeout, s3ReadTimeout)
 	}
+	if transport.DialContext == nil {
+		t.Error("no dialer with a connect timeout")
+	}
+	if !transport.ForceAttemptHTTP2 {
+		t.Error("default options must keep HTTP/2")
+	}
+}
+
+// stallingEndpoint accepts connections and holds them open. With a non-empty
+// head, it answers each request with head and then sends nothing more.
+func stallingEndpoint(t *testing.T, head string) string {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var conns []net.Conn
+	var mu sync.Mutex
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			mu.Lock()
+			conns = append(conns, c)
+			mu.Unlock()
+			if head != "" {
+				go func() {
+					if _, err := http.ReadRequest(bufio.NewReader(c)); err == nil {
+						c.Write([]byte(head))
+					}
+				}()
+			}
+		}
+	}()
+	t.Cleanup(func() {
+		ln.Close()
+		mu.Lock()
+		defer mu.Unlock()
+		for _, c := range conns {
+			c.Close()
+		}
+	})
+	return "http://" + ln.Addr().String()
+}
+
+func readFromStallingEndpoint(t *testing.T, head, wantErr string) {
+	old := s3ReadTimeout
+	s3ReadTimeout = 100 * time.Millisecond
+	defer func() { s3ReadTimeout = old }()
+
+	s, err := newS3BackendStorage(testProperties{
+		"endpoint":              stallingEndpoint(t, head),
+		"region":                "us-east-1",
+		"bucket":                "b",
+		"aws_access_key_id":     "k",
+		"aws_secret_access_key": "s",
+	}, "", "stalling")
+	if err != nil {
+		t.Fatal(err)
+	}
+	f := s.NewStorageFile("key", &volume_server_pb.VolumeInfo{Files: []*volume_server_pb.RemoteFile{{FileSize: 1024}}})
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := f.ReadAt(make([]byte, 8), 0)
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err == nil || !strings.Contains(err.Error(), wantErr) {
+			t.Fatalf("want an error containing %q from a stalled endpoint, got %v", wantErr, err)
+		}
+	case <-time.After(30 * time.Second):
+		t.Fatal("a read from a stalled endpoint did not give up")
+	}
+}
+
+func TestS3BackendReadFailsAgainstSilentEndpoint(t *testing.T) {
+	readFromStallingEndpoint(t, "", "timeout")
+}
+
+func TestS3BackendReadFailsWhenBodyStalls(t *testing.T) {
+	readFromStallingEndpoint(t, "HTTP/1.1 206 Partial Content\r\nContent-Length: 8\r\nContent-Range: bytes 0-7/1024\r\n\r\n", "no data for")
 }
 
 func TestS3BackendSessionsDifferByHTTPOptions(t *testing.T) {

weed/storage/volume_loading.go

@@ -232,6 +232,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		return datFileLoadError(v.FileName(".dat"), err)
 	}
 
+	var remoteSuperBlockErr error
 	if alreadyHasSuperBlock {
 		err = v.readSuperBlock()
 		if err == nil {
@@ -244,6 +245,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		if v.HasRemoteFile() {
 			// maybe temporary network problem
 			logRemoteSuperBlockErr(v.Id, err)
+			remoteSuperBlockErr = err
 			err = nil
 		}
 	} else {
@@ -292,7 +294,9 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		// Recover rows that deletes on a tiered read-only volume overwrote at
 		// the front of .idx. Best effort: a volume that cannot be repaired is
 		// still servable for everything the surviving rows index.
-		if restored, repairErr := v.repairIdxHeadTombstones(); repairErr != nil {
+		if remoteSuperBlockErr != nil {
+			glog.V(1).Infof("volume %d: remote backend unavailable, not recovering overwritten %s rows on this load", v.Id, v.FileName(".idx"))
+		} else if restored, repairErr := v.repairIdxHeadTombstones(); repairErr != nil {
 			glog.Warningf("volume %d: recover overwritten %s rows: %v", v.Id, v.FileName(".idx"), repairErr)
 		} else if restored > 0 {
 			glog.V(0).Infof("volume %d: recovered %d overwritten %s rows from %s", v.Id, restored, v.FileName(".idx"), v.FileName(".dat"))

weed/storage/volume_remote_superblock_stall_test.go

@@ -0,0 +1,185 @@
+package storage
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// stallingBackend serves localDirBackend's objects, but every read waits
+// until release is closed: an endpoint that accepts requests and never answers.
+type stallingBackend struct {
+	*localDirBackend
+	release chan struct{}
+	reads   atomic.Int32
+}
+
+func (b *stallingBackend) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile {
+	return &stallingBackendFile{BackendStorageFile: b.localDirBackend.NewStorageFile(key, tierInfo), backend: b}
+}
+
+type stallingBackendFile struct {
+	backend.BackendStorageFile
+	backend *stallingBackend
+}
+
+func (f *stallingBackendFile) ReadAt(p []byte, off int64) (int, error) {
+	f.backend.reads.Add(1)
+	<-f.backend.release
+	return f.BackendStorageFile.ReadAt(p, off)
+}
+
+func shortenRemoteSuperBlockWait(t *testing.T, timeout, stall time.Duration) {
+	t.Helper()
+	oldTimeout, oldStall := remoteSuperBlockTimeout, remoteSuperBlockStallPeriod
+	remoteSuperBlockTimeout, remoteSuperBlockStallPeriod = timeout, stall
+	remoteSuperBlockStalls.Delete(testBackendName)
+	t.Cleanup(func() {
+		remoteSuperBlockTimeout, remoteSuperBlockStallPeriod = oldTimeout, oldStall
+		remoteSuperBlockStalls.Delete(testBackendName)
+	})
+}
+
+func TestReadRemoteSuperBlockStallSkipsBackend(t *testing.T) {
+	shortenRemoteSuperBlockWait(t, 50*time.Millisecond, time.Hour)
+	b := &stallingBackend{localDirBackend: newLocalDirBackend(t), release: make(chan struct{})}
+	key, _, err := b.CopyFile(writeSuperBlockFile(t), nil, 0)
+	require.NoError(t, err)
+	f := b.NewStorageFile(key, &volume_server_pb.VolumeInfo{Files: []*volume_server_pb.RemoteFile{{FileSize: super_block.SuperBlockSize}}})
+
+	start := time.Now()
+	_, err = readRemoteSuperBlock(testBackendName, f)
+	require.Error(t, err)
+	require.Less(t, time.Since(start), 5*time.Second)
+	require.EqualValues(t, 1, b.reads.Load())
+
+	_, err = readRemoteSuperBlock(testBackendName, f)
+	require.Error(t, err)
+	require.EqualValues(t, 1, b.reads.Load(), "a stalled backend must not be asked again within the stall period")
+
+	close(b.release)
+	remoteSuperBlockStalls.Store(testBackendName, time.Now().Add(-time.Second))
+	sb, err := readRemoteSuperBlock(testBackendName, f)
+	require.NoError(t, err)
+	require.Equal(t, needle.GetCurrentVersion(), sb.Version)
+	_, stalled := remoteSuperBlockStalls.Load(testBackendName)
+	require.False(t, stalled, "a successful read clears the stall")
+}
+
+func writeSuperBlockFile(t *testing.T) *os.File {
+	t.Helper()
+	f, err := os.Create(filepath.Join(t.TempDir(), "sb.dat"))
+	require.NoError(t, err)
+	t.Cleanup(func() { f.Close() })
+	sb := super_block.SuperBlock{Version: needle.GetCurrentVersion(), ReplicaPlacement: &super_block.ReplicaPlacement{}, Ttl: &needle.TTL{}}
+	_, err = f.Write(sb.Bytes())
+	require.NoError(t, err)
+	return f
+}
+
+// A backend that never answers must not hold up the disk scan, local volumes
+// included, and the tiered volumes must serve reads once it answers again.
+func TestRemoteTier_DiskScanDoesNotWaitOnStalledBackend(t *testing.T) {
+	shortenRemoteSuperBlockWait(t, 200*time.Millisecond, time.Hour)
+	local := newLocalDirBackend(t)
+	registerTestBackend(t, local)
+
+	dir := t.TempDir()
+	remoteVids := []needle.VolumeId{61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72}
+	for _, vid := range remoteVids {
+		tierUpVolume(t, dir, vid, local)
+	}
+	lv, err := NewVolume(dir, dir, "", 60, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	lv.Close()
+
+	stalling := &stallingBackend{localDirBackend: local, release: make(chan struct{})}
+	backend.BackendStorages[testBackendName] = stalling
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(stalling.release) }) }
+	t.Cleanup(release)
+
+	loc := &DiskLocation{
+		Directory:              dir,
+		DirectoryUuid:          "test-uuid",
+		IdxDirectory:           dir,
+		DiskType:               types.HddType,
+		MaxVolumeCount:         100,
+		OriginalMaxVolumeCount: 100,
+		MinFreeSpace:           util.MinFreeSpace{Type: util.AsPercent, Percent: 1, Raw: "1"},
+	}
+	loc.volumes = make(map[needle.VolumeId]*Volume)
+	loc.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
+
+	start := time.Now()
+	loc.concurrentLoadingVolumes(NeedleMapInMemory, 2, 0, 0)
+	require.Less(t, time.Since(start), 5*time.Second, "the disk scan waited on the stalled backend")
+	defer func() {
+		for _, v := range loc.volumes {
+			v.Close()
+		}
+	}()
+
+	_, ok := loc.FindVolume(60)
+	require.True(t, ok, "local volume not loaded")
+	for _, vid := range remoteVids {
+		v, ok := loc.FindVolume(vid)
+		require.True(t, ok, "tiered volume %d not loaded", vid)
+		require.True(t, v.HasRemoteFile())
+	}
+	require.Less(t, int(stalling.reads.Load()), len(remoteVids), "every tiered volume waited on the stalled backend")
+
+	release()
+	v, _ := loc.FindVolume(remoteVids[0])
+	_, err = v.readNeedle(newEmptyNeedle(1), nil, nil)
+	require.NoError(t, err, "a tiered volume must serve reads once the backend answers")
+}
+
+// The .idx head repair reads the remote .dat, so it must wait for a load that
+// can reach the backend instead of stalling this one.
+func TestRemoteTier_IdxHeadRepairWaitsForReachableBackend(t *testing.T) {
+	shortenRemoteSuperBlockWait(t, 200*time.Millisecond, time.Hour)
+	local := newLocalDirBackend(t)
+	registerTestBackend(t, local)
+
+	dir := t.TempDir()
+	const vid = needle.VolumeId(80)
+	tierUpVolume(t, dir, vid, local)
+	idxPath := filepath.Join(dir, "80.idx")
+	clobberIdxHead(t, idxPath, []uint64{4, 5})
+	sizeBefore := fileSize(t, idxPath)
+
+	stalling := &stallingBackend{localDirBackend: local, release: make(chan struct{})}
+	backend.BackendStorages[testBackendName] = stalling
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(stalling.release) }) }
+	time.AfterFunc(10*time.Second, release)
+	t.Cleanup(release)
+
+	start := time.Now()
+	v := reloadVolume(t, dir, vid)
+	require.Less(t, time.Since(start), 5*time.Second, "the load waited on the stalled backend")
+	v.Close()
+	require.Equal(t, sizeBefore, fileSize(t, idxPath), "the repair ran against an unreachable backend")
+
+	backend.BackendStorages[testBackendName] = local
+	remoteSuperBlockStalls.Delete(testBackendName)
+	v = reloadVolume(t, dir, vid)
+	defer v.Close()
+	require.Greater(t, fileSize(t, idxPath), sizeBefore, "the repair did not run once the backend answered")
+	_, err := v.readNeedle(newEmptyNeedle(1), nil, nil)
+	require.NoError(t, err)
+}

weed/storage/volume_super_block.go

@@ -3,6 +3,8 @@ package storage
 import (
 	"fmt"
 	"os"
+	"sync"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
@@ -41,7 +43,11 @@ func (v *Volume) maybeWriteSuperBlock(ver needle.Version) error {
 }
 
 func (v *Volume) readSuperBlock() (err error) {
-	v.SuperBlock, err = super_block.ReadSuperBlock(v.DataBackend)
+	if v.HasRemoteFile() && len(v.volumeInfo.GetFiles()) > 0 {
+		v.SuperBlock, err = readRemoteSuperBlock(v.volumeInfo.GetFiles()[0].BackendName(), v.DataBackend)
+	} else {
+		v.SuperBlock, err = super_block.ReadSuperBlock(v.DataBackend)
+	}
 	if v.volumeInfo != nil && v.volumeInfo.Replication != "" {
 		if replication, err := super_block.NewReplicaPlacementFromString(v.volumeInfo.Replication); err != nil {
 			return fmt.Errorf("Error parse volume %d replication %s : %v", v.Id, v.volumeInfo.Replication, err)
@@ -51,3 +57,38 @@ func (v *Volume) readSuperBlock() (err error) {
 	}
 	return err
 }
+
+// A tiered volume loads without its superblock when the backend cannot serve
+// it; once a read stalls, the backend's other volumes skip it for a while.
+var (
+	remoteSuperBlockTimeout     = 10 * time.Second
+	remoteSuperBlockStallPeriod = time.Minute
+	remoteSuperBlockStalls      sync.Map
+)
+
+func readRemoteSuperBlock(backendName string, datBackend backend.BackendStorageFile) (super_block.SuperBlock, error) {
+	if until, ok := remoteSuperBlockStalls.Load(backendName); ok && time.Now().Before(until.(time.Time)) {
+		return super_block.SuperBlock{}, fmt.Errorf("backend %s stalled on a super block read, skipped", backendName)
+	}
+	type result struct {
+		superBlock super_block.SuperBlock
+		err        error
+	}
+	done := make(chan result, 1)
+	go func() {
+		sb, err := super_block.ReadSuperBlock(datBackend)
+		done <- result{sb, err}
+	}()
+	timer := time.NewTimer(remoteSuperBlockTimeout)
+	defer timer.Stop()
+	select {
+	case r := <-done:
+		if r.err == nil {
+			remoteSuperBlockStalls.Delete(backendName)
+		}
+		return r.superBlock, r.err
+	case <-timer.C:
+		remoteSuperBlockStalls.Store(backendName, time.Now().Add(remoteSuperBlockStallPeriod))
+		return super_block.SuperBlock{}, fmt.Errorf("backend %s did not return the super block within %v", backendName, remoteSuperBlockTimeout)
+	}
+}