hyperfile 0.10.2

//! An abort whose log records could not all be deleted.
//!
//! Inside the crate because the faults are injected into the log's own S3 client,
//! which the log builds for itself and nothing outside can reach. Each test swaps
//! the open file's log for one on the same prefix whose client carries an
//! interceptor. They run against S3, like the `integration_s3_*` suites:
//!
//! ```bash
//! cargo test --release --features wal --lib wal::abort_tests -- --ignored --test-threads=1
//! ```

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

use aws_sdk_s3::Client;
use aws_smithy_types::config_bag::{Storable, StoreReplace};

use crate::config::{HyperFileConfig, HyperFileConfigBuilder, HyperFileRuntimeConfig};
use crate::file::flags::{FileFlags, HyperFileFlags};
use crate::file::hyper::Hyper;
use crate::file::mode::{FileMode, HyperFileMode};
use crate::s3uri::S3Uri;
use crate::SegmentId;
use crate::staging::config::StagingConfig;
use crate::wal::config::{HyperFileWalConfig, WalRecoveryMode};
use crate::wal::s3::S3Wal;

const BLK: usize = 4096;

const OFF: usize = 0;
/// Every DeleteObjects is answered 503.
const BUSY: usize = 1;
/// Every key of a DeleteObjects that is a record — not the marker, not the
/// barrier — is refused with AccessDenied, as S3 does when a policy denies some.
const REFUSE_RECORDS: usize = 2;

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

/// The keys of this request that were refused, for the response to list.
#[derive(Debug, Clone)]
struct Refused(Vec<String>);
impl Storable for Refused {
    type Storer = StoreReplace<Self>;
}

fn is_record(key: &str) -> bool {
    !key.ends_with("/txn") && !key.ends_with("/barrier")
}

impl aws_sdk_s3::config::Intercept for Faults {
    fn name(&self) -> &'static str {
        "Faults"
    }
    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> {
        let mode = self.mode.load(Ordering::SeqCst);
        if mode == OFF {
            return Ok(());
        }
        let Some(input) = ctx.input_mut().downcast_mut::<aws_sdk_s3::operation::delete_objects::DeleteObjectsInput>() else {
            return Ok(());
        };
        let mut refused = Vec::new();
        if let Some(d) = input.delete.as_mut() {
            for (n, o) in d.objects.iter_mut().enumerate() {
                if mode == BUSY || is_record(&o.key) {
                    // Sent as a key that does not exist, so S3 deletes nothing
                    // in its place, and the batch is never empty.
                    let stand_in = format!("{}.refused-{}", o.key, n);
                    let real = std::mem::replace(&mut o.key, stand_in);
                    refused.push(real);
                }
            }
        }
        cfg.interceptor_state().store_put(Refused(refused));
        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 Some(Refused(keys)) = cfg.load::<Refused>().cloned() else {
            return Ok(());
        };
        let mode = self.mode.load(Ordering::SeqCst);
        let resp = ctx.response_mut();
        if mode == BUSY {
            *resp.status_mut() = 503u16.try_into().unwrap();
            resp.headers_mut().remove("content-length");
            *resp.body_mut() = aws_smithy_types::body::SdkBody::from(
                "<Error><Code>SlowDown</Code><Message>busy</Message></Error>");
            self.refused.fetch_add(keys.len(), Ordering::SeqCst);
            return Ok(());
        }
        if keys.is_empty() {
            return Ok(());
        }
        self.refused.fetch_add(keys.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 keys {
            xml.push_str(&format!("<Error><Key>{}</Key><Code>AccessDenied</Code><Message>refused by the test</Message></Error>", k));
        }
        xml.push_str("</DeleteResult>");
        resp.headers_mut().remove("content-length");
        *resp.body_mut() = aws_smithy_types::body::SdkBody::from(xml);
        Ok(())
    }
}

fn env(name: &str) -> String {
    std::env::var(name).unwrap_or_else(|_| panic!("{} is not set; these tests need an S3 bucket you own", name))
}

async fn client(faults: Option<&Faults>) -> Client {
    let sdk = aws_config::from_env().region(aws_config::Region::new(env("HYPERFILE_TEST_REGION"))).load().await;
    let mut b = aws_sdk_s3::config::Builder::from(&sdk)
        .retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled());
    if let Some(f) = faults {
        b = b.interceptor(f.clone());
    }
    Client::from_conf(b.build())
}

fn file_uri() -> String {
    format!("s3://{}/hyperfile-test/{}/{}", env("HYPERFILE_TEST_BUCKET"),
        chrono::Utc::now().format("%Y%m%d"), ulid::Ulid::new())
}

fn config(uri: &str, mode: WalRecoveryMode) -> HyperFileConfig {
    let mut runtime = HyperFileRuntimeConfig::default();
    runtime.data_cache_dirty_max_bytes_threshold = usize::MAX / 2;
    runtime.data_cache_dirty_max_blocks_threshold = usize::MAX / 2;
    runtime.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    HyperFileConfigBuilder::new()
        .with_staging_config(&StagingConfig::new_s3_uri(uri, None))
        .with_wal_config(&HyperFileWalConfig::new(&format!("{}/wal", uri)).with_recovery_mode(mode))
        .with_runtime_config(&runtime)
        .build()
}

/// Give `h` a log on the same prefix as its own, whose S3 client is `client`.
/// Called between groups — right after a publish — so the new log starts its
/// numbering where a fresh group does.
fn swap_log(h: &mut Hyper<'static>, uri: &str, client: Client) {
    let wal_uri = format!("{}/wal", uri);
    let s3uri = S3Uri::parse(&wal_uri).expect("uri");
    let root_path = s3uri.key.trim_end_matches('/').to_string();
    h.inner.wal = Some(Box::new(S3Wal {
        client,
        bucket: s3uri.bucket.to_string(),
        root_path_slash: format!("{}/", root_path),
        root_path,
        data_block_size: BLK,
        last_segid: SegmentId::new(0),
        seq: Arc::new(AtomicU64::new(0)),
        pending: Vec::new(),
    }));
}

/// A published block at 0, then a transaction writing at `BLK`, through a log
/// whose client carries `faults`.
async fn in_a_transaction(uri: &str, mode: WalRecoveryMode, faults: &Faults) -> Hyper<'static> {
    let mut h = Hyper::create(client(None).await, config(uri, mode),
        HyperFileFlags::from_flags(FileFlags::rdwr()), HyperFileMode::from_mode(FileMode::default_file()))
        .await.expect("create");
    let _ = h.fs_write(0, &vec![0x31u8; BLK]).await.expect("baseline");
    let _ = h.fs_flush().await.expect("flush");
    swap_log(&mut h, uri, client(Some(faults)).await);
    h.fs_begin_txn().await.expect("begin");
    let _ = h.fs_write(BLK, &vec![0x32u8; BLK]).await.expect("write inside");
    h
}

/// What a later open sees at `BLK`: the bytes read, and whether they are the
/// aborted write's.
async fn reopened(uri: &str, mode: WalRecoveryMode) -> std::io::Result<(usize, bool)> {
    let mut h = Hyper::open(client(None).await, config(uri, mode), HyperFileFlags::from_flags(FileFlags::rdonly())).await?;
    let mut buf = vec![0u8; BLK];
    let n = h.fs_read(BLK, &mut buf).await?;
    let _ = h.fs_release().await;
    Ok((n, n > 0 && buf.iter().all(|b| *b == 0x32)))
}

async fn remove(uri: &str) {
    let _ = Hyper::fs_unlink(&client(None).await, uri).await;
}

/// An abort that could not delete the log fails, leaves the transaction open, and
/// can be asked again. It closed the transaction first, so a second abort was
/// refused as "no transaction is open" with the records still there.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore]
async fn an_abort_that_could_not_clear_the_log_can_be_asked_again() {
    let _ = env_logger::try_init();
    let uri = file_uri();
    let faults = Faults::default();
    let mut h = in_a_transaction(&uri, WalRecoveryMode::Latest, &faults).await;

    faults.mode.store(BUSY, Ordering::SeqCst);
    let first = h.fs_abort_txn().await;
    faults.mode.store(OFF, Ordering::SeqCst);
    let still_open = h.fs_in_txn();
    let again = h.fs_abort_txn().await;
    let closed = !h.fs_in_txn();
    let _ = h.fs_release().await;
    let seen = reopened(&uri, WalRecoveryMode::Latest).await;
    remove(&uri).await;

    let e = first.expect_err("the abort succeeded with the log not deleted");
    assert_eq!(e.kind(), ErrorKind::ResourceBusy, "{}", e);
    assert!(faults.refused.load(Ordering::SeqCst) > 0, "nothing was refused: the test did not test");
    assert!(still_open, "a failed abort must leave the transaction open, so it can be asked again");
    again.expect("the second abort");
    assert!(closed);
    let (_, came_back) = seen.expect("reopen");
    assert!(!came_back, "the aborted write came back on reopen");
}

/// An abort refused part-way never brings the aborted writes back, even if it
/// is not asked again: the records are deleted before the marker, so a marker
/// left behind still keeps them out. The marker went in the same batch as the
/// records, so S3 could delete it and refuse them, and the next open — finding
/// unsealed records and no marker — replayed them in `Latest`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore]
async fn an_abort_refused_part_way_never_brings_the_writes_back() {
    let _ = env_logger::try_init();
    for mode in [WalRecoveryMode::Latest, WalRecoveryMode::Barrier] {
        let uri = file_uri();
        let faults = Faults::default();
        let mut h = in_a_transaction(&uri, mode, &faults).await;

        faults.mode.store(REFUSE_RECORDS, Ordering::SeqCst);
        let first = h.fs_abort_txn().await;
        faults.mode.store(OFF, Ordering::SeqCst);
        // Gone without asking again, as a process that dies would be.
        std::mem::forget(h);
        let seen = reopened(&uri, mode).await;
        remove(&uri).await;

        let e = first.expect_err("the abort succeeded with records refused");
        assert_eq!(e.kind(), ErrorKind::PermissionDenied, "{:?}: {}", mode, e);
        assert!(faults.refused.load(Ordering::SeqCst) > 0, "{:?}: nothing was refused", mode);
        let (_, came_back) = seen.unwrap_or_else(|e| panic!("{:?}: reopen: {}", mode, e));
        assert!(!came_back, "{:?}: the aborted write came back on reopen", mode);
    }
}