hyperfile 0.10.2

//! The error kinds a caller decides retries by.
//!
//! What hyperfile returns follows one rule (see `s3_error_kind`): `ConnectionReset`,
//! `TimedOut` and `ResourceBusy` are worth retrying, `StaleNetworkFileHandle` means
//! reopen, the rest are not worth retrying. These pin the two cases that broke it:
//!
//! - a GetObject whose body is cut off after the response headers came: the SDK's
//!   own conversion makes that `Other`, though it is the network;
//! - a store answering 503 to a reactor write: the handler read `ResourceBusy` as
//!   "requeued, answered later", so the caller waited for an answer that never
//!   came — or, through a stopgap, got `Other`.
//!
//! And that a requeue the handler makes itself still never reaches the caller.
//!
//! ```bash
//! cargo test --release --test integration_s3_error_kinds -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use std::io::ErrorKind;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};

use hyperfile::file::flags::FileFlags;
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;

const BS: usize = 4096;

/// A body that yields some bytes and then fails, as a connection dropped part-way.
struct CutOff(Option<bytes::Bytes>);

impl http_body::Body for CutOff {
    type Data = bytes::Bytes;
    type Error = std::io::Error;
    fn poll_data(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Result<Self::Data, Self::Error>>> {
        match self.0.take() {
            Some(b) => Poll::Ready(Some(Ok(b))),
            None => Poll::Ready(Some(Err(std::io::Error::new(ErrorKind::UnexpectedEof, "connection dropped by the test")))),
        }
    }
    fn poll_trailers(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
        Poll::Ready(Ok(None))
    }
}

/// While armed: ranged GetObject responses — data, not the inode — have their
/// body cut off after the first bytes.
#[derive(Debug, Clone, Default)]
struct Faults {
    cut_bodies: Arc<AtomicBool>,
}

impl aws_sdk_s3::config::Intercept for Faults {
    fn name(&self) -> &'static str {
        "Faults"
    }
    fn modify_before_deserialization(
        &self,
        ctx: &mut aws_sdk_s3::config::interceptors::BeforeDeserializationInterceptorContextMut<'_>,
        _rc: &aws_sdk_s3::config::RuntimeComponents,
        _cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> Result<(), aws_sdk_s3::error::BoxError> {
        if self.cut_bodies.load(Ordering::SeqCst) && ctx.response().status().is_success() {
            let resp = ctx.response_mut();
            let is_get_of_data = resp.headers().get("content-range").is_some();
            if is_get_of_data {
                let first = bytes::Bytes::from_static(&[0u8; 100]);
                *resp.body_mut() = aws_smithy_types::body::SdkBody::from_body_0_4(CutOff(Some(first)));
            }
        }
        Ok(())
    }
}

async fn faulty_client() -> (aws_sdk_s3::Client, Faults) {
    let faults = Faults::default();
    let sdk = aws_config::from_env().region(aws_config::Region::new(test_region())).load().await;
    // No retries of its own, so what is seen is the first failure's kind.
    let conf = aws_sdk_s3::config::Builder::from(&sdk)
        .retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled())
        .interceptor(faults.clone()).build();
    (aws_sdk_s3::Client::from_conf(conf), faults)
}

async fn seeded(client: &aws_sdk_s3::Client) -> TestFile {
    let tf = TestFile::new(client).await;
    let mut h = Hyper::fs_open_or_create_with_default_opt(client, tf.uri(), FileFlags::rdwr(), FileMode::default_file())
        .await.expect("create");
    h.fs_write(0, &vec![0x5au8; 64 * BS]).await.expect("write");
    h.fs_flush().await.expect("flush");
    let _ = h.fs_release().await;
    tf
}

/// A data read whose body is cut off part-way is `ConnectionReset`, and the same
/// read succeeds once the network is back.
#[tokio::test]
#[ignore]
async fn a_body_cut_off_part_way_is_connection_reset() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = seeded(&client).await;
    let (faulty, faults) = faulty_client().await;
    let mut h = Hyper::fs_open(&faulty, tf.uri(), FileFlags::rdonly()).await.expect("open");
    let mut buf = vec![0u8; 64 * BS];
    faults.cut_bodies.store(true, Ordering::SeqCst);
    let cut = h.fs_read(0, &mut buf).await;
    faults.cut_bodies.store(false, Ordering::SeqCst);
    let again = h.fs_read(0, &mut buf).await;
    let _ = h.fs_release().await;
    tf.cleanup(&client).await;

    let e = cut.expect_err("the read with its body cut off succeeded");
    assert_eq!(e.kind(), ErrorKind::ConnectionReset, "{}", e);
    assert_eq!(again.expect("the read once the network is back"), 64 * BS);
    assert!(buf.iter().all(|b| *b == 0x5a));
}

/// A store answering 503 to a reactor write's block fetch reaches the caller as
/// `ResourceBusy` — answered, not taken for a requeue — and the handler serves on.
#[cfg(feature = "reactor")]
#[tokio::test]
#[ignore]
async fn a_busy_store_reaches_a_reactor_caller_as_resource_busy() {
    use hyperfile::file::fh::HyperFileHandler;
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = seeded(&client).await;

    #[derive(Debug, Clone)]
    struct Busy(Arc<AtomicBool>);
    impl aws_sdk_s3::config::Intercept for Busy {
        fn name(&self) -> &'static str { "Busy" }
        fn modify_before_deserialization(
            &self,
            ctx: &mut aws_sdk_s3::config::interceptors::BeforeDeserializationInterceptorContextMut<'_>,
            _rc: &aws_sdk_s3::config::RuntimeComponents,
            _cfg: &mut aws_sdk_s3::config::ConfigBag,
        ) -> Result<(), aws_sdk_s3::error::BoxError> {
            if self.0.load(Ordering::SeqCst) && ctx.response().headers().get("content-range").is_some() {
                let resp: &mut aws_sdk_s3::config::http::HttpResponse = ctx.response_mut();
                *resp.status_mut() = 503u16.try_into().unwrap();
                resp.headers_mut().remove("content-range");
                resp.headers_mut().remove("content-length");
                *resp.body_mut() = aws_smithy_types::body::SdkBody::from(
                    "<Error><Code>SlowDown</Code><Message>busy</Message></Error>");
            }
            Ok(())
        }
    }
    let busy = Busy(Arc::new(AtomicBool::new(false)));
    let sdk = aws_config::from_env().region(aws_config::Region::new(test_region())).load().await;
    let conf = aws_sdk_s3::config::Builder::from(&sdk)
        .retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled())
        .interceptor(busy.clone()).build();
    let busy_client = aws_sdk_s3::Client::from_conf(conf);

    let reactor = hyperfile_reactor::Reactor::new_current().expect("reactor");
    let mut fh = HyperFileHandler::fh_open(&reactor, &busy_client, tf.uri(), FileFlags::rdwr()).await.expect("fh_open");
    busy.0.store(true, Ordering::SeqCst);
    // A partial write: its block has to be fetched first.
    let failed = tokio::time::timeout(std::time::Duration::from_secs(30), fh.fh_write(100, b"partial")).await;
    busy.0.store(false, Ordering::SeqCst);
    let again = fh.fh_write(100, b"partial").await;
    let flushed = fh.fh_flush().await;
    let _ = fh.fh_release().await;
    drop(fh);
    tf.cleanup(&client).await;

    let failed = failed.expect("the write was never answered");
    let e = failed.expect_err("the write succeeded with the store busy");
    assert_eq!(e.kind(), ErrorKind::ResourceBusy, "{}", e);
    again.expect("the write once the store answers");
    flushed.expect("flush");
}

/// Requeues the handler makes itself still never reach the caller: writes racing
/// a flush are put back and answered once it is done.
#[cfg(feature = "reactor")]
#[tokio::test]
#[ignore]
async fn a_requeue_is_never_seen_by_the_caller() {
    use hyperfile::file::fh::HyperFileHandler;
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = seeded(&client).await;
    let reactor = hyperfile_reactor::Reactor::new_current().expect("reactor");
    let fh = HyperFileHandler::fh_open(&reactor, &client, tf.uri(), FileFlags::rdwr()).await.expect("fh_open");
    let mut tasks = Vec::new();
    for i in 0..16usize {
        let mut f = fh.clone();
        tasks.push(tokio::spawn(async move {
            let w = f.fh_write(i * BS + 7, &[i as u8; 100]).await;
            let fl = if i % 4 == 0 { Some(f.fh_flush().await) } else { None };
            (w, fl)
        }));
    }
    let mut results = Vec::new();
    for t in tasks {
        results.push(tokio::time::timeout(std::time::Duration::from_secs(60), t).await);
    }
    let mut fh = fh;
    let _ = fh.fh_release().await;
    drop(fh);
    tf.cleanup(&client).await;
    for (i, r) in results.into_iter().enumerate() {
        let (w, fl) = r.expect("an operation was never answered").expect("task");
        w.unwrap_or_else(|e| panic!("write {} saw {:?}: {}", i, e.kind(), e));
        if let Some(fl) = fl {
            fl.unwrap_or_else(|e| panic!("flush {} saw {:?}: {}", i, e.kind(), e));
        }
    }
}