use std::path::{Path, PathBuf};
use std::sync::Arc;
use azure_core::credentials::TokenCredential;
use azure_core::http::HttpClient;
use cbh_config::{CloudStorageConfig, Config};
use cbh_diag::{Reporter, ReporterExt};
use cbh_model::sanitize_segment;
use super::azure::AzureBlobStorage;
use super::caching::CachingStorage;
use super::local::LocalStorage;
use super::{Storage, StorageError};
fn rebase(base: &Path, path: PathBuf) -> PathBuf {
if path.is_absolute() {
path
} else {
base.join(path)
}
}
#[derive(Clone, Debug)]
pub enum StorageFacade {
Local(LocalStorage),
Azure(AzureBlobStorage),
CachedAzure(CachingStorage<AzureBlobStorage, LocalStorage>),
}
impl Storage for StorageFacade {
#[cfg_attr(coverage_nightly, coverage(off))]
async fn put(&self, key: &str, bytes: &[u8]) -> Result<(), StorageError> {
match self {
Self::Local(storage) => storage.put(key, bytes).await,
Self::Azure(storage) => storage.put(key, bytes).await,
Self::CachedAzure(_) => {
unreachable!("a cached backend is never selected for an object-writing command")
}
}
}
#[cfg_attr(coverage_nightly, coverage(off))]
async fn put_overwrite(&self, key: &str, bytes: &[u8]) -> Result<(), StorageError> {
match self {
Self::Local(storage) => storage.put_overwrite(key, bytes).await,
Self::Azure(storage) => storage.put_overwrite(key, bytes).await,
Self::CachedAzure(_) => {
unreachable!("a cached backend is never selected for an object-writing command")
}
}
}
async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError> {
match self {
Self::Local(storage) => storage.get(key).await,
Self::Azure(storage) => storage.get(key).await,
Self::CachedAzure(storage) => storage.get(key).await,
}
}
async fn list(&self, prefix: &str) -> Result<Vec<String>, StorageError> {
match self {
Self::Local(storage) => storage.list(prefix).await,
Self::Azure(storage) => storage.list(prefix).await,
Self::CachedAzure(storage) => storage.list(prefix).await,
}
}
async fn delete(&self, key: &str) -> Result<(), StorageError> {
match self {
Self::Local(storage) => storage.delete(key).await,
Self::Azure(storage) => storage.delete(key).await,
Self::CachedAzure(storage) => storage.delete(key).await,
}
}
}
impl StorageFacade {
fn cached_at(self, mirror_root: PathBuf) -> Self {
match self {
Self::Azure(backend) => {
Self::CachedAzure(CachingStorage::new(backend, LocalStorage::new(mirror_root)))
}
other @ (Self::Local(_) | Self::CachedAzure(_)) => other,
}
}
pub async fn synchronize_cache(
&self,
project: &str,
reporter: &dyn Reporter,
) -> Result<(), StorageError> {
match self {
Self::Local(_) | Self::Azure(_) => Ok(()),
Self::CachedAzure(storage) => storage.synchronize(project, reporter).await,
}
}
pub async fn flush_pending_invalidation(
&self,
project: &str,
reporter: &dyn Reporter,
) -> Result<(), StorageError> {
match self {
Self::Local(_) => Ok(()),
Self::Azure(storage) => storage.flush_pending_invalidation(project, reporter).await,
Self::CachedAzure(storage) => {
storage
.inner()
.flush_pending_invalidation(project, reporter)
.await
}
}
}
pub fn report_cache_tally(&self, reporter: &dyn Reporter) {
if let Self::CachedAzure(storage) = self {
storage.report_tally(reporter);
}
}
}
pub fn build_storage(
local: Option<&Path>,
config: &Config,
base: &Path,
cache: Option<&Path>,
) -> Result<StorageFacade, StorageError> {
if let Some(path) = local {
return Ok(StorageFacade::Local(LocalStorage::new(rebase(
base,
path.to_path_buf(),
))));
}
match &config.storage {
Some(CloudStorageConfig::Azure(azure)) => {
let backend = AzureBlobStorage::from_config(
&azure.account,
&azure.container,
azure.endpoint.clone(),
)?;
match cache {
Some(cache_dir) => {
let mirror_root = rebase(base, cache_dir.to_path_buf())
.join(sanitize_segment(&azure.account))
.join(sanitize_segment(&azure.container));
Ok(StorageFacade::CachedAzure(CachingStorage::new(
backend,
LocalStorage::new(mirror_root),
)))
}
None => Ok(StorageFacade::Azure(backend)),
}
}
None => Err(StorageError::Config {
message: "no storage configured: pass --local=<path> (or set \
CARGO_BENCH_HISTORY_STORAGE and pass a bare --local) or \
configure a cloud storage backend in the configuration file"
.to_owned(),
}),
}
}
pub fn resolve_storage(
storage_override: Option<StorageFacade>,
local: Option<&Path>,
config: &Config,
base: &Path,
cache: Option<&Path>,
reporter: &dyn Reporter,
) -> Result<StorageFacade, StorageError> {
match storage_override {
Some(backend) => {
reporter.note_with(|| "storage backend: injected by test override".to_owned());
Ok(match cache {
Some(cache_dir) => {
if matches!(backend, StorageFacade::Azure(_)) {
reporter.note_with(|| {
"--cache is set: wrapping the injected Azure backend in a \
read-through cache mirror so the override honors --cache like \
a configured backend does"
.to_owned()
});
} else {
reporter.note_with(|| {
"--cache is set, but the injected backend is not a plain Azure \
backend, so it is used unchanged — a read-through cache only \
fronts a cloud backend"
.to_owned()
});
}
backend.cached_at(rebase(base, cache_dir.to_path_buf()))
}
None => backend,
})
}
None => build_storage(local, config, base, cache),
}
}
#[doc(hidden)]
#[derive(Clone, Debug)]
pub struct StorageOverride(pub(crate) StorageFacade);
impl StorageOverride {
#[must_use]
pub fn into_facade(self) -> StorageFacade {
self.0
}
}
#[doc(hidden)]
pub fn azure_backend_from_parts(
account: &str,
container: &str,
endpoint: Option<String>,
credential: Arc<dyn TokenCredential>,
http_client: Arc<dyn HttpClient>,
) -> Result<StorageOverride, StorageError> {
let backend =
AzureBlobStorage::from_parts(account, container, endpoint, credential, http_client)?;
Ok(StorageOverride(StorageFacade::Azure(backend)))
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
use cbh_config::parse_config;
use cbh_diag::RecordingReporter;
use tempfile::tempdir;
use super::*;
fn config_with_storage(storage: &str) -> Config {
parse_config(storage).unwrap()
}
#[tokio::test]
#[cfg_attr(miri, ignore)] async fn local_facade_dispatches_every_storage_method() {
let dir = tempdir().unwrap();
let storage = StorageFacade::Local(LocalStorage::new(dir.path()));
storage.put("v1/proj/run.json", b"first").await.unwrap();
assert_eq!(storage.get("v1/proj/run.json").await.unwrap(), b"first");
storage
.put_overwrite("v1/proj/run.json", b"second")
.await
.unwrap();
assert_eq!(storage.get("v1/proj/run.json").await.unwrap(), b"second");
let keys = storage.list("v1/proj/").await.unwrap();
assert_eq!(keys, vec!["v1/proj/run.json".to_owned()]);
storage.delete("v1/proj/run.json").await.unwrap();
let error = storage.get("v1/proj/run.json").await.unwrap_err();
assert!(matches!(error, StorageError::NotFound { .. }), "{error:?}");
}
#[test]
fn build_storage_for_local_yields_a_local_backend() {
let config = Config::default();
let storage =
build_storage(Some(Path::new("./data")), &config, Path::new("/work"), None).unwrap();
assert!(matches!(storage, StorageFacade::Local(_)));
assert!(format!("{storage:?}").contains("data"), "{storage:?}");
}
#[test]
fn build_storage_local_overrides_a_configured_cloud_backend() {
let config = config_with_storage("[storage.azure]\naccount = \"a\"\ncontainer = \"c\"\n");
let storage =
build_storage(Some(Path::new("./data")), &config, Path::new("/work"), None).unwrap();
assert!(matches!(storage, StorageFacade::Local(_)), "{storage:?}");
}
#[test]
fn build_storage_without_selection_or_config_is_a_config_error() {
let config = Config::default();
let error = build_storage(None, &config, Path::new("/work"), None).unwrap_err();
assert!(matches!(error, StorageError::Config { .. }), "{error:?}");
}
#[test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
fn build_storage_for_azure_yields_an_azure_backend() {
let config = config_with_storage(
"[storage.azure]\naccount = \"devstoreaccount1\"\ncontainer = \"bench-history\"\n\
endpoint = \"https://devstoreaccount1.blob.core.windows.net\"\n",
);
let storage = build_storage(None, &config, Path::new("/work"), None).unwrap();
assert!(matches!(storage, StorageFacade::Azure(_)));
}
#[test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
fn build_storage_for_azure_with_a_cache_yields_a_cached_backend() {
let config = config_with_storage(
"[storage.azure]\naccount = \"devstoreaccount1\"\ncontainer = \"bench-history\"\n\
endpoint = \"https://devstoreaccount1.blob.core.windows.net\"\n",
);
let storage = build_storage(
None,
&config,
Path::new("/work"),
Some(Path::new("./cache")),
)
.unwrap();
assert!(
matches!(storage, StorageFacade::CachedAzure(_)),
"{storage:?}"
);
}
#[test]
fn build_storage_local_ignores_a_cache_directory() {
let config = Config::default();
let storage = build_storage(
Some(Path::new("./data")),
&config,
Path::new("/work"),
Some(Path::new("./cache")),
)
.unwrap();
assert!(matches!(storage, StorageFacade::Local(_)), "{storage:?}");
}
fn azure_cache_facade() -> StorageFacade {
let config = config_with_storage(
"[storage.azure]\naccount = \"devstoreaccount1\"\ncontainer = \"bench-history\"\n\
endpoint = \"https://devstoreaccount1.blob.core.windows.net\"\n",
);
build_storage(
None,
&config,
Path::new("/work"),
Some(Path::new("./cache")),
)
.unwrap()
}
#[test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
fn report_cache_tally_notes_the_tally_for_a_cached_backend() {
let storage = azure_cache_facade();
let reporter = RecordingReporter::new();
storage.report_cache_tally(&reporter);
assert!(
reporter.contains("served 0 objects from the local mirror and fetched 0 objects"),
"{:?}",
reporter.notes()
);
}
#[tokio::test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
#[cfg_attr(
mutants,
ignore = "A PendingInvalidation::take mutant turns this no-op into a real marker write against the configured Azure endpoint, which has no reachable backend here; the client's retry/backoff trips the mutation timeout. take()'s contract is caught fast by pending_invalidation_starts_unarmed."
)]
async fn flush_pending_invalidation_for_a_cached_backend_without_mutations_is_a_noop() {
let storage = azure_cache_facade();
let reporter = RecordingReporter::new();
storage
.flush_pending_invalidation("proj", &reporter)
.await
.unwrap();
assert!(reporter.notes().is_empty(), "{:?}", reporter.notes());
}
fn plain_azure_facade() -> StorageFacade {
let config = config_with_storage(
"[storage.azure]\naccount = \"devstoreaccount1\"\ncontainer = \"bench-history\"\n\
endpoint = \"https://devstoreaccount1.blob.core.windows.net\"\n",
);
build_storage(None, &config, Path::new("/work"), None).unwrap()
}
#[test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
fn build_storage_namespaces_the_cache_mirror_by_backend_identity() {
fn mirror_root(account: &str, container: &str) -> String {
let config = config_with_storage(&format!(
"[storage.azure]\naccount = \"{account}\"\ncontainer = \"{container}\"\n\
endpoint = \"https://{account}.blob.core.windows.net\"\n"
));
let storage = build_storage(
None,
&config,
Path::new("/work"),
Some(Path::new("./cache")),
)
.unwrap();
assert!(
matches!(storage, StorageFacade::CachedAzure(_)),
"{storage:?}"
);
let rendered = format!("{storage:?}");
rendered
.split_once("root: \"")
.expect("the cache backend renders its root")
.1
.split_once('"')
.expect("the root path is quoted")
.0
.to_owned()
}
let one = mirror_root("alpha", "first");
let two = mirror_root("beta", "second");
assert!(
one.contains("alpha") && one.contains("first"),
"the mirror root is namespaced by backend identity: {one}"
);
assert!(
two.contains("beta") && two.contains("second"),
"the mirror root is namespaced by backend identity: {two}"
);
assert_ne!(
one, two,
"distinct Azure backends sharing a --cache dir must get distinct mirror roots"
);
}
#[test]
fn resolve_storage_without_an_override_builds_from_config() {
let reporter = RecordingReporter::new();
let storage = resolve_storage(
None,
Some(Path::new("./data")),
&Config::default(),
Path::new("/work"),
None,
&reporter,
)
.unwrap();
assert!(matches!(storage, StorageFacade::Local(_)), "{storage:?}");
assert!(
!reporter.contains("injected by test override"),
"{:?}",
reporter.notes()
);
}
#[test]
fn resolve_storage_uses_an_uncached_override_verbatim() {
let injected = StorageFacade::Local(LocalStorage::new(Path::new("/mirror")));
let reporter = RecordingReporter::new();
let storage = resolve_storage(
Some(injected),
None,
&Config::default(),
Path::new("/work"),
None,
&reporter,
)
.unwrap();
assert!(matches!(storage, StorageFacade::Local(_)), "{storage:?}");
assert!(
reporter.contains("injected by test override"),
"{:?}",
reporter.notes()
);
assert!(
!reporter.contains("read-through cache"),
"{:?}",
reporter.notes()
);
}
#[test]
fn resolve_storage_leaves_a_non_azure_override_unchanged_under_cache() {
let injected = StorageFacade::Local(LocalStorage::new(Path::new("/mirror")));
let reporter = RecordingReporter::new();
let storage = resolve_storage(
Some(injected),
None,
&Config::default(),
Path::new("/work"),
Some(Path::new("/cache")),
&reporter,
)
.unwrap();
assert!(matches!(storage, StorageFacade::Local(_)), "{storage:?}");
assert!(
reporter.contains("used unchanged") && !reporter.contains("wrapping"),
"{:?}",
reporter.notes()
);
}
#[test]
#[cfg_attr(
miri,
ignore = "builds a real HTTP client (reqwest) that Miri cannot run"
)]
fn resolve_storage_wraps_an_azure_override_to_honor_cache() {
let injected = plain_azure_facade();
assert!(matches!(injected, StorageFacade::Azure(_)), "{injected:?}");
let cache = tempdir().unwrap();
let reporter = RecordingReporter::new();
let storage = resolve_storage(
Some(injected),
None,
&Config::default(),
Path::new("/work"),
Some(cache.path()),
&reporter,
)
.unwrap();
assert!(
matches!(storage, StorageFacade::CachedAzure(_)),
"{storage:?}"
);
assert!(
reporter.contains("injected by test override")
&& reporter.contains("wrapping the injected Azure backend"),
"{:?}",
reporter.notes()
);
}
}