hyperfile 0.10.2

//! A write whose index lookup fails part-way returns an error and changes nothing.
//!
//! Every write path inserts into the block map for each block it touches, and an
//! insert may have to read an index node back from S3. A read that failed — a
//! network drop, say — used to panic the process (`bmap.insert(..).expect(..)`), and
//! returning the error alone would not have been enough: the blocks before the
//! failure were already in the map and the cache, the ones after only in the cache,
//! and the size was not yet updated, so the next flush either failed for good or
//! stored pointers its index never recorded.
//!
//! A write now changes the block map first and undoes what it changed if a later
//! block fails, and changes the cache, references and inode only once every block
//! is in the map. These tests make an index read fail in the middle of a write and
//! check, for each write path: the write returns an error, the file reads and stats
//! as it did before, the same write repeated succeeds, and what is stored then is
//! what a twin file that never saw the failure stores.
//!
//! How the failure lands mid-write: the file holds two regions written in separate
//! flushes, R1 first and then R0 below it. R1's leaves past its first stay in R1's
//! segment; everything else the index needs for R0 is in R0's segment. The test
//! reads R0, which brings its index into memory, then refuses every read of R1's
//! segment and writes a range from R0's tail through a hole into R1. The blocks in
//! R0 and the hole go into the map; the first block whose leaf is in R1's segment
//! fails.
//!
//! ```bash
//! cargo test --release --test integration_s3_write_failure -- --ignored --test-threads=1
//! ```

// The WAL build refuses `write_ref`, and leaves what only that test uses unused.
#![cfg_attr(feature = "wal", allow(unused_imports, dead_code))]

mod common;
use common::*;

use std::io::{Error, ErrorKind, Result};
use std::sync::{Arc, Mutex};

use btree_ondisk::NullNodeCache;
#[cfg(not(feature = "wal"))]
use hyperfile::block_source::BlockSource;
use hyperfile::buffer::{AlignedDataBlockWrapper, BatchDataBlockWrapper};
use hyperfile::config::{HyperFileConfigBuilder, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;
use hyperfile::file::HyperTrait;
use hyperfile::meta_loader::s3::S3BlockLoader;
use hyperfile::segment::Segment;
use hyperfile::staging::{config::StagingConfig, s3::S3Staging, Staging};
use hyperfile::SegmentId;

type S3File = HyperFile<'static, S3Staging, S3BlockLoader, NullNodeCache>;

const BS: usize = 4096;
const R0: usize = 0;
const R0_LEN: usize = 2048;
const R1: usize = 2304;
const R1_LEN: usize = 2048;
/// From R0's tail, through the hole between the regions, well into R1: past the
/// leaf R1 shares with R0, so a leaf only R1's segment holds is reached.
const OP_FIRST: usize = R0 + R0_LEN - 8;
const OP_END: usize = R1 + 900;
const END: usize = R1 + R1_LEN;

fn tag(blk: usize) -> u8 {
    (blk % 251) as u8 + 1
}

fn region(first: usize, len: usize) -> Vec<u8> {
    let mut v = vec![0u8; len * BS];
    for (i, b) in v.chunks_mut(BS).enumerate() {
        b.fill(tag(first + i));
        b[..8].copy_from_slice(&((first + i) as u64).to_le_bytes());
    }
    v
}

/// What the file holds before the operation: R0, a hole, R1.
fn before() -> Vec<u8> {
    let mut v = region(R0, R0_LEN);
    v.resize(R1 * BS, 0);
    v.extend_from_slice(&region(R1, R1_LEN));
    v
}

/// Refuses every GetObject of one key while armed.
#[derive(Debug, Clone, Default)]
struct RefuseReads(Arc<Mutex<Option<String>>>);

impl RefuseReads {
    fn arm(&self, key_suffix: &str) {
        *self.0.lock().unwrap() = Some(key_suffix.to_string());
    }
    fn disarm(&self) {
        *self.0.lock().unwrap() = None;
    }
}

impl aws_sdk_s3::config::Intercept for RefuseReads {
    fn name(&self) -> &'static str {
        "RefuseReads"
    }
    fn read_before_execution(
        &self,
        ctx: &aws_sdk_s3::config::interceptors::BeforeSerializationInterceptorContextRef<'_>,
        _cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> std::result::Result<(), aws_sdk_s3::error::BoxError> {
        if let Some(input) = ctx.input().downcast_ref::<aws_sdk_s3::operation::get_object::GetObjectInput>() {
            if let (Some(armed), Some(key)) = (self.0.lock().unwrap().as_ref(), input.key()) {
                if key.ends_with(armed.as_str()) {
                    return Err(format!("read of {} refused by the test", key).into());
                }
            }
        }
        Ok(())
    }
}

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

fn runtime() -> HyperFileRuntimeConfig {
    let mut r = HyperFileRuntimeConfig::default();
    r.data_cache_dirty_max_bytes_threshold = 1 << 40;
    r.data_cache_dirty_max_blocks_threshold = (1 << 40) / BS;
    r.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    r.segment_buffer_size = 1 << 30;
    r
}

/// A file holding R1 then R0, each in its own flush. Returns the name of R1's
/// segment, whose reads the test refuses.
async fn prepared(client: &aws_sdk_s3::Client) -> (TestFile, String) {
    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(R1 * BS, &region(R1, R1_LEN)).await.expect("write R1");
    let r1_cno = h.fs_flush().await.expect("flush R1");
    h.fs_write(R0 * BS, &region(R0, R0_LEN)).await.expect("write R0");
    h.fs_flush().await.expect("flush R0");
    let _ = h.fs_release().await;
    let r1 = Segment::segid_to_staging_file_id(SegmentId::new_from_cno(r1_cno).at_part(0));
    (tf, r1)
}

async fn read_all(h: &mut Hyper<'static>) -> Vec<u8> {
    let mut v = vec![0u8; END * BS];
    let n = h.fs_read(0, &mut v).await.expect("read");
    v.truncate(n);
    v
}

/// Size, blocks and bytes, read from a fresh handle.
async fn stored(client: &aws_sdk_s3::Client, uri: &str) -> (i64, i64, Vec<u8>) {
    let mut h = Hyper::fs_open(client, uri, FileFlags::rdonly()).await.expect("open to read");
    let st = h.fs_getattr().expect("getattr");
    let data = read_all(&mut h).await;
    let _ = h.fs_release().await;
    (st.st_size, st.st_blocks, data)
}

/// The operations under test, on a `Hyper` handle.
#[derive(Clone, Copy, Debug)]
enum Op {
    Write,
    WriteZero,
    WriteBatch,
    WriteAlignedBatch,
}

/// What the file holds after `op` succeeds.
fn after(op: Op) -> Vec<u8> {
    let mut v = before();
    let (s, e) = (OP_FIRST * BS, OP_END * BS);
    match op {
        Op::Write => {
            // starts 100 bytes into the first block
            for (i, b) in v[s + 100..e].iter_mut().enumerate() { *b = (i % 253) as u8 ^ 0x5a; }
        }
        Op::WriteZero => v[s + 100..e].fill(0),
        Op::WriteBatch | Op::WriteAlignedBatch => {
            for blk in OP_FIRST..OP_END {
                let b = &mut v[blk * BS..(blk + 1) * BS];
                if blk % 7 == 3 { b.fill(0) } else { b.fill(0xc0 ^ tag(blk)) }
            }
            if let Op::WriteBatch = op {
                // and a partial overwrite of the first block, 200 bytes at 300
                v[s + 300..s + 500].fill(0xee);
            }
        }
    }
    v
}

async fn apply(h: &mut Hyper<'static>, op: Op) -> Result<usize> {
    let (s, e) = (OP_FIRST * BS, OP_END * BS);
    match op {
        Op::Write => {
            let buf: Vec<u8> = (0..e - s - 100).map(|i| (i % 253) as u8 ^ 0x5a).collect();
            h.fs_write(s + 100, &buf).await
        }
        Op::WriteZero => h.fs_write_zero(s + 100, e - s - 100).await,
        Op::WriteBatch => {
            let mut blocks = Vec::new();
            for blk in OP_FIRST..OP_END {
                let zero = blk % 7 == 3;
                let b = BatchDataBlockWrapper::new(blk as u64, BS, zero);
                if !zero { b.as_mut_slice().fill(0xc0 ^ tag(blk)); }
                blocks.push(b);
            }
            let p = BatchDataBlockWrapper::new_partial_block(OP_FIRST as u64, BS, 300, 200, false);
            p.as_mut_slice().fill(0xee);
            blocks.push(p);
            h.fs_write_batch(blocks).await
        }
        Op::WriteAlignedBatch => {
            let mut blocks = Vec::new();
            for blk in OP_FIRST..OP_END {
                let zero = blk % 7 == 3;
                let b = AlignedDataBlockWrapper::new(blk as u64, BS, zero);
                if !zero { b.as_mut_slice().fill(0xc0 ^ tag(blk)); }
                blocks.push(b);
            }
            h.fs_write_aligned_batch(blocks).await
        }
    }
}

/// The whole check for one operation on a `Hyper` handle.
async fn fails_cleanly_and_retries(op: Op) {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let (tf, r1_seg) = prepared(&client).await;
    let (twin, _) = prepared(&client).await;
    let (refusing, refuse) = refusing_client().await;

    // the twin never sees a failure
    {
        let mut h = Hyper::fs_open_opt(&client, twin.uri(), FileFlags::rdwr(), &runtime()).await.expect("open twin");
        apply(&mut h, op).await.expect("the twin's write");
        h.fs_flush().await.expect("the twin's flush");
        let _ = h.fs_release().await;
    }

    let mut h = Hyper::fs_open_opt(&refusing, tf.uri(), FileFlags::rdwr(), &runtime()).await.expect("open");
    let st0 = h.fs_getattr().expect("getattr");
    let mut r0 = vec![0u8; R0_LEN * BS];
    h.fs_read(R0 * BS, &mut r0).await.expect("read R0, bringing its index in");
    refuse.arm(&r1_seg);
    let failed = apply(&mut h, op).await;
    refuse.disarm();

    let st1 = h.fs_getattr().expect("getattr after the failure");
    let seen = read_all(&mut h).await;
    let retried = apply(&mut h, op).await;
    let flushed = h.fs_flush().await;
    let _ = h.fs_release().await;
    let mine = stored(&client, tf.uri()).await;
    let theirs = stored(&client, twin.uri()).await;
    tf.cleanup(&client).await;
    twin.cleanup(&client).await;

    assert!(failed.is_err(), "{:?}: the write succeeded, so the failure did not land: {:?}", op, failed);
    assert_eq!((st1.st_size, st1.st_blocks), (st0.st_size, st0.st_blocks), "{:?}: the failed write changed size or blocks", op);
    assert!(seen == before(), "{:?}: after the failed write the file does not read as before", op);
    retried.expect("the same write, repeated");
    flushed.expect("flush after the repeat");
    assert!(theirs.2 == after(op), "{:?}: the twin's write did not store what was written", op);
    assert!(mine.2 == theirs.2, "{:?}: what is stored differs from the twin's", op);
    assert_eq!((mine.0, mine.1), (theirs.0, theirs.1), "{:?}: size or blocks differ from the twin's", op);
}

#[tokio::test]
#[ignore]
async fn a_write_that_fails_part_way_changes_nothing_and_can_be_repeated() {
    fails_cleanly_and_retries(Op::Write).await;
}

#[tokio::test]
#[ignore]
async fn a_write_zero_that_fails_part_way_changes_nothing_and_can_be_repeated() {
    fails_cleanly_and_retries(Op::WriteZero).await;
}

#[tokio::test]
#[ignore]
async fn a_write_batch_that_fails_part_way_changes_nothing_and_can_be_repeated() {
    fails_cleanly_and_retries(Op::WriteBatch).await;
}

#[tokio::test]
#[ignore]
async fn an_aligned_batch_that_fails_part_way_changes_nothing_and_can_be_repeated() {
    fails_cleanly_and_retries(Op::WriteAlignedBatch).await;
}

#[cfg(not(feature = "wal"))]
/// A source over the bytes the reference stands for.
struct Src(Vec<u8>);

#[cfg(not(feature = "wal"))]
impl BlockSource for Src {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
        let o = off as usize;
        if o + buf.len() > self.0.len() {
            return Err(Error::new(ErrorKind::UnexpectedEof, "past the source"));
        }
        buf.copy_from_slice(&self.0[o..o + buf.len()]);
        Ok(())
    }
}

#[cfg(not(feature = "wal"))]
async fn s3file(client: &aws_sdk_s3::Client, uri: &str) -> S3File {
    let rt = runtime();
    let sc = StagingConfig::new_s3_uri(uri, None);
    let config = HyperFileConfigBuilder::new().with_staging_config(&sc).with_runtime_config(&rt).build();
    let staging = S3Staging::from(client, config.staging.clone(), config.runtime.clone()).await.expect("staging");
    let loader = staging.to_block_loader();
    S3File::open(staging, loader, NullNodeCache, config, HyperFileFlags::from_flags(FileFlags::rdwr()))
        .await.expect("open")
}

/// `write_ref` over the same range: references are recorded only once every block
/// is in the map. Not with the WAL, which refuses `write_ref`.
#[cfg(not(feature = "wal"))]
#[tokio::test]
#[ignore]
async fn a_write_ref_that_fails_part_way_changes_nothing_and_can_be_repeated() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let (tf, r1_seg) = prepared(&client).await;
    let (twin, _) = prepared(&client).await;
    let (refusing, refuse) = refusing_client().await;
    let mut want = before();
    let src: Vec<u8> = (0..END * BS).map(|i| (i / BS) as u8 ^ 0x33).collect();
    want[OP_FIRST * BS..OP_END * BS].copy_from_slice(&src[OP_FIRST * BS..OP_END * BS]);

    {
        let mut f = s3file(&client, twin.uri()).await;
        f.set_block_source(Some(Arc::new(Src(src.clone())))).unwrap();
        f.write_ref(OP_FIRST * BS, (OP_END - OP_FIRST) * BS).await.expect("the twin's write_ref");
        f.flush().await.expect("the twin's flush");
    }

    let mut f = s3file(&refusing, tf.uri()).await;
    f.set_block_source(Some(Arc::new(Src(src.clone())))).unwrap();
    let blocks0 = f.stat().st_blocks;
    let mut r0 = vec![0u8; R0_LEN * BS];
    f.read(R0 * BS, &mut r0).await.expect("read R0, bringing its index in");
    refuse.arm(&r1_seg);
    let failed = f.write_ref(OP_FIRST * BS, (OP_END - OP_FIRST) * BS).await;
    refuse.disarm();
    let refs_after_failure = f.refs_len();
    let blocks1 = f.stat().st_blocks;
    let mut seen = vec![0u8; END * BS];
    let n = f.read(0, &mut seen).await.expect("read");
    seen.truncate(n);
    let retried = f.write_ref(OP_FIRST * BS, (OP_END - OP_FIRST) * BS).await;
    let flushed = f.flush().await;
    drop(f);
    let mine = stored(&client, tf.uri()).await;
    let theirs = stored(&client, twin.uri()).await;
    tf.cleanup(&client).await;
    twin.cleanup(&client).await;

    assert!(failed.is_err(), "the write_ref succeeded, so the failure did not land");
    assert_eq!(refs_after_failure, 0, "a failed write_ref left references");
    assert_eq!(blocks1, blocks0, "a failed write_ref changed the block count");
    assert!(seen == before(), "after the failed write_ref the file does not read as before");
    retried.expect("the same write_ref, repeated");
    flushed.expect("flush after the repeat");
    assert!(theirs.2 == want, "the twin did not store what was referenced");
    assert!(mine.2 == theirs.2, "what is stored differs from the twin's");
    assert_eq!((mine.0, mine.1), (theirs.0, theirs.1), "size or blocks differ from the twin's");
}

/// Through the reactor: the write is absorbed on the handler task, and its error
/// comes back to the caller; the handler goes on serving.
#[cfg(feature = "reactor")]
#[tokio::test]
#[ignore]
async fn a_reactor_write_that_fails_part_way_returns_the_error_and_can_be_repeated() {
    use hyperfile::file::fh::HyperFileHandler;
    use hyperfile_reactor::Reactor;
    let _ = env_logger::try_init();
    let client = make_client().await;
    let (tf, r1_seg) = prepared(&client).await;
    let (refusing, refuse) = refusing_client().await;
    let reactor = Reactor::new_current().expect("reactor");
    let mut fh = HyperFileHandler::fh_open(&reactor, &refusing, tf.uri(), FileFlags::rdwr()).await.expect("fh_open");
    let mut r0 = vec![0u8; R0_LEN * BS];
    fh.fh_read(R0 * BS, &mut r0).await.expect("read R0, bringing its index in");
    let buf: Vec<u8> = (0..(OP_END - OP_FIRST) * BS).map(|i| (i % 253) as u8 ^ 0x5a).collect();
    refuse.arm(&r1_seg);
    let failed = fh.fh_write(OP_FIRST * BS, &buf).await;
    refuse.disarm();
    let st = fh.fh_getattr().await;
    let mut seen = vec![0u8; END * BS];
    let n = fh.fh_read(0, &mut seen).await.expect("read after the failure");
    seen.truncate(n);
    let retried = fh.fh_write(OP_FIRST * BS, &buf).await;
    let flushed = fh.fh_flush().await;
    let _ = fh.fh_release().await;
    drop(fh);
    let mine = stored(&client, tf.uri()).await;
    tf.cleanup(&client).await;

    let mut want = before();
    want[OP_FIRST * BS..OP_END * BS].copy_from_slice(&buf);
    assert!(failed.is_err(), "the write succeeded, so the failure did not land");
    st.expect("the handler serves after the failure");
    assert!(seen == before(), "after the failed write the file does not read as before");
    retried.expect("the same write, repeated");
    flushed.expect("flush after the repeat");
    assert!(mine.2 == want, "what is stored is not what was written");
}