hyperfile 0.10.4

//! A DeleteObjects that succeeds without deleting everything.
//!
//! The response lists each key it could not delete — a policy denying some
//! prefixes, a throttled partition — and is 200 all the same. `unlink` read only
//! the status, so it reported success with objects left behind, after deleting
//! the inode: objects nothing could find again. An SDK interceptor plays S3's
//! part here: it takes the refused keys out of the request and lists them as
//! errors in the response, as S3 does.
//!
//! ```bash
//! cargo test --release --test integration_s3_partial_delete -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use std::io::ErrorKind;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

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

const BS: usize = 4096;

/// While armed, every DeleteObjects has all its keys but the first refused, with
/// `code`.
#[derive(Debug, Clone)]
struct RefuseKeys {
    armed: Arc<AtomicBool>,
    code: &'static str,
    held: Arc<Mutex<Vec<String>>>,
    refused: Arc<AtomicUsize>,
}

impl RefuseKeys {
    fn new(code: &'static str) -> Self {
        Self { armed: Arc::new(AtomicBool::new(false)), code, held: Arc::default(), refused: Arc::default() }
    }
}

impl aws_sdk_s3::config::Intercept for RefuseKeys {
    fn name(&self) -> &'static str {
        "RefuseKeys"
    }
    fn modify_before_serialization(
        &self,
        ctx: &mut aws_sdk_s3::config::interceptors::BeforeSerializationInterceptorContextMut<'_>,
        _rc: &aws_sdk_s3::config::RuntimeComponents,
        _cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> Result<(), aws_sdk_s3::error::BoxError> {
        if !self.armed.load(Ordering::SeqCst) {
            return Ok(());
        }
        if let Some(input) = ctx.input_mut().downcast_mut::<aws_sdk_s3::operation::delete_objects::DeleteObjectsInput>() {
            if let Some(d) = input.delete.as_mut() {
                if d.objects.len() > 1 {
                    let refused = d.objects.split_off(1);
                    self.held.lock().unwrap().extend(refused.into_iter().map(|o| o.key));
                }
            }
        }
        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> {
        let held: Vec<String> = std::mem::take(&mut *self.held.lock().unwrap());
        if held.is_empty() || !ctx.response().status().is_success() {
            return Ok(());
        }
        self.refused.fetch_add(held.len(), Ordering::SeqCst);
        let mut xml = String::from("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
            <DeleteResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">");
        for k in held {
            xml.push_str(&format!("<Error><Key>{}</Key><Code>{}</Code><Message>refused by the test</Message></Error>", k, self.code));
        }
        xml.push_str("</DeleteResult>");
        let resp = ctx.response_mut();
        resp.headers_mut().remove("content-length");
        *resp.body_mut() = aws_smithy_types::body::SdkBody::from(xml);
        Ok(())
    }
}

async fn refusing_client(refuse: &RefuseKeys) -> aws_sdk_s3::Client {
    let sdk = aws_config::from_env().region(aws_config::Region::new(test_region())).load().await;
    aws_sdk_s3::Client::from_conf(aws_sdk_s3::config::Builder::from(&sdk).interceptor(refuse.clone()).build())
}

/// A file with several objects under its prefix: the inode and a few segments.
async fn file_with_segments(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");
    for i in 0..3usize {
        h.fs_write(i * BS, &vec![i as u8 + 1; BS]).await.expect("write");
        h.fs_flush().await.expect("flush");
    }
    let _ = h.fs_release().await;
    tf
}

async fn keys_under(client: &aws_sdk_s3::Client, uri: &str) -> usize {
    let u = uri.trim_start_matches("s3://");
    let (bucket, prefix) = u.split_once('/').unwrap();
    client.list_objects_v2().bucket(bucket).prefix(format!("{}/", prefix.trim_end_matches('/')))
        .send().await.expect("list").key_count().unwrap_or(0) as usize
}

/// Delete everything under the file's prefix, without the inode `unlink` needs.
async fn delete_prefix(client: &aws_sdk_s3::Client, uri: &str) {
    let u = uri.trim_start_matches("s3://");
    let (bucket, prefix) = u.split_once('/').unwrap();
    let list = client.list_objects_v2().bucket(bucket).prefix(format!("{}/", prefix.trim_end_matches('/')))
        .send().await.expect("list");
    for o in list.contents() {
        let _ = client.delete_object().bucket(bucket).key(o.key().unwrap()).send().await;
    }
}

/// `unlink` fails when S3 refuses some of the keys, with the kind of the refusal,
/// and the message names the code.
///
/// It fails after the inode is gone, so running it again does not finish the job —
/// see "Known gap: orphaned objects" in `docs/posix.md`. What changes is that the
/// caller is told, and the log says which keys.
#[tokio::test]
#[ignore]
async fn unlink_fails_when_some_keys_are_not_deleted() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let refuse = RefuseKeys::new("AccessDenied");
    let faulty = refusing_client(&refuse).await;
    let tf = file_with_segments(&client).await;
    let before = keys_under(&client, tf.uri()).await;

    refuse.armed.store(true, Ordering::SeqCst);
    let first = Hyper::fs_unlink(&faulty, tf.uri()).await;
    refuse.armed.store(false, Ordering::SeqCst);
    let left = keys_under(&client, tf.uri()).await;
    delete_prefix(&client, tf.uri()).await;

    assert!(before >= 3, "the file should have several objects, has {}", before);
    assert!(refuse.refused.load(Ordering::SeqCst) > 0, "nothing was refused: the test did not test");
    assert!(left > 0, "the refused keys should still be there");
    let e = first.expect_err("unlink reported success with objects left behind");
    assert_eq!(e.kind(), ErrorKind::PermissionDenied, "{}", e);
    assert!(e.to_string().contains("AccessDenied"), "the message should name the code: {}", e);
}

/// The first refusal's code decides the kind: a throttled key is `ResourceBusy`,
/// the kind worth retrying.
#[tokio::test]
#[ignore]
async fn a_key_refused_as_slow_down_is_resource_busy() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let refuse = RefuseKeys::new("SlowDown");
    let faulty = refusing_client(&refuse).await;
    let tf = file_with_segments(&client).await;

    refuse.armed.store(true, Ordering::SeqCst);
    let first = Hyper::fs_unlink(&faulty, tf.uri()).await;
    refuse.armed.store(false, Ordering::SeqCst);
    delete_prefix(&client, tf.uri()).await;

    let e = first.expect_err("unlink reported success with objects left behind");
    assert_eq!(e.kind(), ErrorKind::ResourceBusy, "{}", e);
}

/// An unlink that S3 lets delete everything still succeeds.
#[tokio::test]
#[ignore]
async fn unlink_with_nothing_refused_succeeds() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let refuse = RefuseKeys::new("AccessDenied");
    let faulty = refusing_client(&refuse).await;
    let tf = file_with_segments(&client).await;
    let r = Hyper::fs_unlink(&faulty, tf.uri()).await;
    let after = keys_under(&client, tf.uri()).await;
    if after > 0 {
        delete_prefix(&client, tf.uri()).await;
    }
    r.expect("unlink");
    assert_eq!(after, 0);
}