use std::time::Duration;
use aws_sdk_s3::Client;
use uuid::Uuid;
use hyperdir::{HyperDirLayout, ROOT_DIR_UUID};
use hyperdir::hyper::HyperDir;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::mode::FileMode;
use hyperfile::ondisk::InodeRaw;
fn bucket() -> String {
std::env::var("S3_BUCKET").expect("S3_BUCKET not set (source ../hyperdir-env.sh)")
}
async fn make_client() -> Client {
let region = std::env::var("S3_REGION").expect("S3_REGION not set");
let config = aws_config::from_env().region(aws_config::Region::new(region)).load().await;
Client::new(&config)
}
fn dir_mode() -> FileMode { FileMode::from(0o755) }
fn file_mode() -> FileMode { FileMode::from(0o644) }
fn file_inode(size: usize) -> InodeRaw {
InodeRaw {
i_mode: 0o100_644, i_nlink: 1,
i_size: size as u64,
..Default::default()
}
}
fn io_err<E: std::fmt::Display>(e: E) -> std::io::Error {
std::io::Error::other(e.to_string())
}
async fn object_exists(client: &Client, bucket: &str, key: &str) -> bool {
client.head_object().bucket(bucket).key(key).send().await.is_ok()
}
async fn count_suffix(client: &Client, bucket: &str, prefix: &str, suffix: &str) -> usize {
let mut stream = client.list_objects_v2().bucket(bucket).prefix(prefix).into_paginator().send();
let mut n = 0;
while let Some(page) = stream.next().await {
let page = page.expect("list");
for o in page.contents() {
if o.key().map(|k| k.ends_with(suffix)).unwrap_or(false) { n += 1; }
}
}
n
}
async fn purge_prefix(client: &Client, bucket: &str, prefix: &str) {
let mut stream = client.list_objects_v2().bucket(bucket).prefix(prefix).into_paginator().send();
let mut keys: Vec<String> = Vec::new();
while let Some(page) = stream.next().await {
let page = page.expect("list for cleanup");
for o in page.contents() { if let Some(k) = o.key() { keys.push(k.to_string()); } }
}
for chunk in keys.chunks(1000) {
let ids: Vec<_> = chunk.iter().map(|k| {
aws_sdk_s3::types::ObjectIdentifier::builder().key(k).build().unwrap()
}).collect();
let del = aws_sdk_s3::types::Delete::builder().set_objects(Some(ids)).build().unwrap();
let _ = client.delete_objects().bucket(bucket).delete(del).send().await;
}
}
#[tokio::test]
#[ignore]
async fn e2e_blob_lifecycle() {
let client = make_client().await;
let bucket = bucket();
let base = format!("hyperdir-e2e-blob/{}", Uuid::new_v4());
let layout = HyperDirLayout::with_base(&base);
let result = run(&client, &bucket, &layout).await;
purge_prefix(&client, &bucket, &format!("{}/", base)).await;
result.expect("e2e blob flow");
}
async fn run(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
let data1 = vec![0xABu8; 200];
let (f_uuid, ulid1) = HyperDir::fs_create_blob(
client, layout, bucket, &ROOT_DIR_UUID, "f", &file_inode(data1.len()), &data1).await?;
let e = HyperDir::fs_resolve_entry_fast(client, layout, bucket, &ROOT_DIR_UUID, "f").await?
.expect("f resolves");
assert_eq!(e.uuid, f_uuid);
assert_eq!(e.blob_ulid, Some(ulid1), "entry points at v1");
assert_eq!(e.inline, None, "a blob is not in-inode inline");
assert_eq!(HyperDir::fs_read_blob(client, layout, bucket, &f_uuid, ulid1).await?, data1,
"one GET returns v1 bytes");
let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
let _ = root.fs_compact().await?;
drop(root);
let e = HyperDir::fs_resolve_entry_fast(client, layout, bucket, &ROOT_DIR_UUID, "f").await?
.expect("f after compact");
assert_eq!(e.blob_ulid, Some(ulid1), "blob_ulid preserved through compaction");
let data2 = vec![0xCDu8; 300];
let ulid2 = HyperDir::fs_write_blob(
client, layout, bucket, &ROOT_DIR_UUID, "f", &f_uuid, &file_inode(data2.len()), &data2).await?;
assert_ne!(ulid2, ulid1);
let e = HyperDir::fs_resolve_entry_fast(client, layout, bucket, &ROOT_DIR_UUID, "f").await?.unwrap();
assert_eq!(e.blob_ulid, Some(ulid2), "entry now points at v2");
assert_eq!(HyperDir::fs_read_blob(client, layout, bucket, &f_uuid, ulid2).await?, data2);
assert_eq!(HyperDir::fs_read_blob(client, layout, bucket, &f_uuid, ulid1).await?, data1,
"old version still present (retention/undelete window)");
let dropped = HyperDir::fs_gc_blob_versions(client, layout, bucket, &f_uuid, ulid2, Duration::from_millis(0)).await?;
assert_eq!(dropped, 1, "exactly the superseded v1 reclaimed");
assert!(HyperDir::fs_read_blob(client, layout, bucket, &f_uuid, ulid1).await.is_err(), "v1 gone");
assert_eq!(HyperDir::fs_read_blob(client, layout, bucket, &f_uuid, ulid2).await?, data2, "v2 kept");
let _h = HyperDir::fs_spill_blob(
client, layout, bucket, &ROOT_DIR_UUID, "f", &f_uuid, ulid2, FileFlags::rdwr(), file_mode()).await?;
let e = HyperDir::fs_resolve_entry_fast(client, layout, bucket, &ROOT_DIR_UUID, "f").await?.unwrap();
assert_eq!(e.blob_ulid, None, "f is now segmented (commit flipped blob_ulid to 0)");
let f_prefix = format!("{}/", layout.file_key(&f_uuid));
assert_eq!(count_suffix(client, bucket, &f_prefix, ".blob").await, 0, "blob versions cleaned after spill");
assert!(object_exists(client, bucket, &format!("{}inode", f_prefix)).await, "segmented inode exists");
let stray_key = layout.blob_key(&f_uuid, &ulid1.to_string());
client.put_object().bucket(bucket).key(&stray_key)
.body(aws_sdk_s3::primitives::ByteStream::from(vec![0u8; 16]))
.send().await.map_err(io_err)?;
assert!(object_exists(client, bucket, &stray_key).await, "stray planted");
let _ = HyperDir::fs_gc_orphans(client, layout, bucket, Duration::from_millis(0)).await?;
assert!(!object_exists(client, bucket, &stray_key).await, "stray blob under segmented file reclaimed (Gap B)");
assert!(object_exists(client, bucket, &format!("{}inode", f_prefix)).await, "segmented file left intact");
let data3 = vec![0x11u8; 500];
let (g_uuid, _g_ulid) = HyperDir::fs_create_blob(
client, layout, bucket, &ROOT_DIR_UUID, "g", &file_inode(data3.len()), &data3).await?;
let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
let _ = root.fs_compact().await?;
drop(root);
HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "g", &g_uuid, false, Some(Duration::from_millis(0))).await?;
let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
let _ = root.fs_compact().await?; drop(root);
assert!(HyperDir::fs_resolve_entry_fast(client, layout, bucket, &ROOT_DIR_UUID, "g").await?.is_none(),
"g unlinked");
let g_prefix = format!("{}/", layout.file_key(&g_uuid));
assert!(count_suffix(client, bucket, &g_prefix, ".blob").await >= 1, "g blob still present before sweep");
let _ = HyperDir::fs_gc_orphans(client, layout, bucket, Duration::from_millis(0)).await?;
assert_eq!(count_suffix(client, bucket, &g_prefix, ".blob").await, 0,
"deleted blob reclaimed by the orphan sweep");
Ok(())
}