hyperfile 0.10.3

//! Input that is not what hyperfile expects is an error, never a panic or a hang.
//!
//! - A URI that does not name a bucket and a path is `InvalidInput`, from open,
//!   create and unlink alike. Several panicked on an `unwrap` of the parse.
//! - A HeadObject answered without a usable `Date` or `Last-Modified` — a proxy, an
//!   S3-compatible store — is `InvalidData`. The server's time is what a writable
//!   open uses to decide whether a checkpoint no inode points at yet is still being
//!   published; an unparseable `Date` panicked, a missing one left the time at 0,
//!   so the checkpoint looked forever fresh and open waited for good.
//!
//! ```bash
//! cargo test --release --test integration_s3_bad_input -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use aws_smithy_types::config_bag::{Storable, StoreReplace};
use futures::FutureExt;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;

type Caught<T> = Result<std::io::Result<T>, String>;

async fn caught<T>(f: impl std::future::Future<Output = std::io::Result<T>>) -> Caught<T> {
    std::panic::AssertUnwindSafe(f).catch_unwind().await.map_err(|p| {
        p.downcast_ref::<String>().cloned()
            .or_else(|| p.downcast_ref::<&str>().map(|s| s.to_string()))
            .unwrap_or_default()
    })
}

fn kind<T>(r: &Caught<T>) -> String {
    match r {
        Err(p) => format!("panic: {}", p),
        Ok(Ok(_)) => "ok".to_string(),
        Ok(Err(e)) => format!("{:?}", e.kind()),
    }
}

/// Each URI that does not name a bucket and a path is refused as `InvalidInput`,
/// without a panic, by every entry point that takes one.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore]
async fn a_uri_without_a_bucket_and_a_path_is_invalid_input() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let b = test_bucket();
    let uris = ["s3://".to_string(), "s3:///".to_string(), "s3:///x".to_string(),
        format!("s3://{}", b), format!("s3://{}/", b), "http://x/y".to_string(), String::new()];
    let mut wrong = Vec::new();
    for u in &uris {
        let open = caught(Hyper::fs_open(&client, u, FileFlags::rdonly())).await;
        let create = caught(Hyper::fs_open_or_create_with_default_opt(&client, u, FileFlags::rdwr(),
            FileMode::default_file())).await;
        let unlink = caught(Hyper::fs_unlink(&client, u)).await;
        for (what, k) in [("open", kind(&open)), ("create", kind(&create)), ("unlink", kind(&unlink))] {
            if k != "InvalidInput" {
                wrong.push(format!("{:?} {}: {}", u.replace(&b, "<bucket>"), what, k));
            }
        }
    }
    assert!(wrong.is_empty(), "not InvalidInput:\n{}", wrong.join("\n"));
}

/// The message says which URI and why.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore]
async fn the_refusal_names_the_uri() {
    let client = make_client().await;
    let e = Hyper::fs_open(&client, "s3://", FileFlags::rdonly()).await.err().expect("refused");
    assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput);
    assert!(e.to_string().contains("\"s3://\"") && e.to_string().contains("Incomplete"), "{}", e);
}

const KEEP: usize = 0;
const GARBLE_DATE: usize = 1;
const DROP_DATE: usize = 2;
const DROP_LAST_MODIFIED: usize = 3;

#[derive(Debug, Clone, Default)]
struct HeadFaults {
    mode: Arc<AtomicUsize>,
    hit: Arc<AtomicUsize>,
}

#[derive(Debug, Clone)]
struct IsHead;
impl Storable for IsHead {
    type Storer = StoreReplace<Self>;
}

impl aws_sdk_s3::config::Intercept for HeadFaults {
    fn name(&self) -> &'static str {
        "HeadFaults"
    }
    fn read_before_execution(
        &self,
        ctx: &aws_sdk_s3::config::interceptors::BeforeSerializationInterceptorContextRef<'_>,
        cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> Result<(), aws_sdk_s3::error::BoxError> {
        if ctx.input().downcast_ref::<aws_sdk_s3::operation::head_object::HeadObjectInput>().is_some() {
            cfg.interceptor_state().store_put(IsHead);
        }
        Ok(())
    }
    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 cfg.load::<IsHead>().is_none() || !ctx.response().status().is_success() {
            return Ok(());
        }
        let h = ctx.response_mut().headers_mut();
        match self.mode.load(Ordering::SeqCst) {
            GARBLE_DATE => { h.insert("date", "not a date"); }
            DROP_DATE => { h.remove("date"); }
            DROP_LAST_MODIFIED => { h.remove("last-modified"); }
            _ => return Ok(()),
        }
        self.hit.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

/// A container whose last flush wrote its segment and not its inode: the next
/// writable open has to judge from the server's time whether that checkpoint is
/// still being published.
async fn with_a_pending_checkpoint(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, b"published").await.unwrap();
    h.fs_flush().await.unwrap();
    h.with_staging_interceptor(FailOnFlushInode::from(1));
    h.fs_write(0, b"left pending").await.unwrap();
    let _ = h.fs_flush().await.expect_err("every inode write is refused");
    let _ = h.fs_release().await;
    tf
}

/// A HeadObject without a usable `Date` or `Last-Modified` makes a writable open
/// over a pending checkpoint fail with `InvalidData`, promptly, and an ordinary
/// answer lets the same open take the checkpoint up.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore]
async fn a_head_without_a_usable_date_is_invalid_data_not_a_panic_or_a_wait() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let faults = HeadFaults::default();
    let sdk = aws_config::from_env().region(aws_config::Region::new(test_region())).load().await;
    let faulty = aws_sdk_s3::Client::from_conf(aws_sdk_s3::config::Builder::from(&sdk)
        .interceptor(faults.clone()).build());
    let mut wrong = Vec::new();
    for (name, mode) in [("garbled Date", GARBLE_DATE), ("no Date", DROP_DATE), ("no Last-Modified", DROP_LAST_MODIFIED)] {
        let tf = with_a_pending_checkpoint(&client).await;
        faults.mode.store(mode, Ordering::SeqCst);
        let before = faults.hit.load(Ordering::SeqCst);
        let opened = tokio::time::timeout(Duration::from_secs(20),
            caught(async { Hyper::fs_open(&faulty, tf.uri(), FileFlags::rdwr()).await.map(|_| ()) })).await;
        let hit = faults.hit.load(Ordering::SeqCst) > before;
        faults.mode.store(KEEP, Ordering::SeqCst);
        let mut buf = vec![0u8; 12];
        let fine = match Hyper::fs_open(&faulty, tf.uri(), FileFlags::rdwr()).await {
            Ok(mut h) => { let n = h.fs_read(0, &mut buf).await.unwrap_or(0); let _ = h.fs_release().await; n == 12 && &buf == b"left pending" }
            Err(_) => false,
        };
        tf.cleanup(&client).await;
        let got = match &opened { Err(_) => "no answer in 20 s".to_string(), Ok(r) => kind(r) };
        if !hit { wrong.push(format!("{}: no HeadObject was altered; the test did not test", name)); }
        if got != "InvalidData" { wrong.push(format!("{}: {}", name, got)); }
        if !fine { wrong.push(format!("{}: the open with ordinary answers did not take the checkpoint up", name)); }
    }
    assert!(wrong.is_empty(), "\n{}", wrong.join("\n"));
}