use std::sync::Arc;
use async_trait::async_trait;
use bytes::Bytes;
use crate::{
gate, BootstrapReport, ComposeSpec, DeleteManyReport, EpochLease, EpochState,
GenerationStamps, LifecycleView,
ListedObject, ListedVersion, ObjectMeta, ObjectStore, PendingUpload, PresignedPut, PutCondition,
RequestCounts, RetentionOutcome, StoreError, StoreResult,
};
pub struct ReadOnly<S: ?Sized> {
inner: Arc<S>,
}
impl<S: ObjectStore + ?Sized> ReadOnly<S> {
pub fn new(inner: Arc<S>) -> Self {
ReadOnly { inner }
}
}
fn refuse<T>(verb: &str, key: &str) -> StoreResult<T> {
Err(StoreError::Auth(format!("read-only store: {verb} {key}")))
}
#[async_trait]
impl<S: ObjectStore + ?Sized> ObjectStore for ReadOnly<S> {
async fn put_whole(
&self,
key: &str,
_body: Bytes,
_condition: &PutCondition,
_stamps: &GenerationStamps,
_crc64: u64,
) -> StoreResult<ObjectMeta> {
refuse("put_whole", key)
}
async fn copy_object(
&self,
_src_key: &str,
_src_if_match: Option<&str>,
dst_key: &str,
_condition: &PutCondition,
_stamps: &GenerationStamps,
) -> StoreResult<ObjectMeta> {
refuse("copy_object", dst_key)
}
async fn compose_generation(&self, spec: &ComposeSpec<'_>) -> StoreResult<ObjectMeta> {
refuse("compose_generation", spec.key)
}
async fn delete(&self, key: &str) -> StoreResult<()> {
refuse("delete", key)
}
async fn delete_if_match(&self, key: &str, _etag: &str) -> StoreResult<()> {
refuse("delete_if_match", key)
}
async fn delete_many(&self, keys: &[String]) -> StoreResult<DeleteManyReport> {
refuse("delete_many", keys.first().map(String::as_str).unwrap_or("(no keys)"))
}
async fn delete_version(&self, key: &str, _version_id: &str) -> StoreResult<()> {
refuse("delete_version", key)
}
async fn presign_put(
&self,
key: &str,
_ttl_secs: u64,
_sha256: &[u8; 32],
) -> StoreResult<PresignedPut> {
refuse("presign_put", key)
}
async fn ensure_noncurrent_retention(
&self,
prefix: &str,
_days: u64,
) -> StoreResult<RetentionOutcome> {
refuse("ensure_noncurrent_retention", prefix)
}
async fn abort_upload(&self, key: &str, _upload_id: &str) -> StoreResult<()> {
refuse("abort_upload", key)
}
async fn bootstrap(&self, prefix: &str) -> StoreResult<BootstrapReport> {
refuse("bootstrap", prefix)
}
async fn epoch_acquire(
&self,
key: &str,
_holder_id: &str,
_supersede: Option<&EpochState>,
) -> StoreResult<EpochLease> {
refuse("epoch_acquire", key)
}
async fn epoch_renew(
&self,
key: &str,
_lease: &EpochLease,
_echo: Option<&str>,
) -> StoreResult<EpochLease> {
refuse("epoch_renew", key)
}
async fn epoch_release(&self, key: &str, _lease: &EpochLease) -> StoreResult<()> {
refuse("epoch_release", key)
}
async fn epoch_enqueue(
&self,
key: &str,
_observed: &EpochState,
_holder_id: &str,
) -> StoreResult<EpochState> {
refuse("epoch_enqueue", key)
}
async fn epoch_handoff(
&self,
key: &str,
_lease: &EpochLease,
_echo: Option<&str>,
) -> StoreResult<()> {
refuse("epoch_handoff", key)
}
async fn head(&self, key: &str) -> StoreResult<ObjectMeta> {
self.inner.head(key).await
}
async fn get_whole(&self, key: &str, if_match: Option<&str>) -> StoreResult<(ObjectMeta, Bytes)> {
self.inner.get_whole(key, if_match).await
}
async fn get_range(
&self,
key: &str,
offset: u64,
len: u64,
if_match: &str,
) -> StoreResult<Bytes> {
self.inner.get_range(key, offset, len, if_match).await
}
async fn get_range_segments(
&self,
key: &str,
offset: u64,
len: u64,
if_match: &str,
) -> StoreResult<Vec<Bytes>> {
self.inner.get_range_segments(key, offset, len, if_match).await
}
async fn list(&self, prefix: &str) -> StoreResult<Vec<ListedObject>> {
self.inner.list(prefix).await
}
async fn head_version(&self, key: &str, version_id: &str) -> StoreResult<ObjectMeta> {
self.inner.head_version(key, version_id).await
}
async fn get_version(&self, key: &str, version_id: &str) -> StoreResult<(ObjectMeta, Bytes)> {
self.inner.get_version(key, version_id).await
}
async fn list_versions(&self, prefix: &str) -> StoreResult<Vec<ListedVersion>> {
self.inner.list_versions(prefix).await
}
async fn presign_get(&self, key: &str, ttl_secs: u64) -> StoreResult<String> {
self.inner.presign_get(key, ttl_secs).await
}
async fn lifecycle_rules(&self) -> StoreResult<Vec<LifecycleView>> {
self.inner.lifecycle_rules().await
}
async fn list_uploads(&self, prefix: &str) -> StoreResult<Vec<PendingUpload>> {
self.inner.list_uploads(prefix).await
}
async fn epoch_read(&self, key: &str) -> StoreResult<Option<EpochState>> {
self.inner.epoch_read(key).await
}
fn min_part_size(&self) -> u64 {
self.inner.min_part_size()
}
fn max_parts(&self) -> usize {
self.inner.max_parts()
}
fn upload_gate(&self) -> Option<Arc<gate::ByteGate>> {
self.inner.upload_gate()
}
fn request_counts(&self) -> Option<RequestCounts> {
self.inner.request_counts()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::memory::MemoryStore;
use crate::PartSource;
use std::sync::Mutex;
const WRITES: &[&str] = &[
"put_whole", "copy_object", "compose_generation", "delete", "delete_if_match",
"delete_many", "delete_version", "presign_put", "ensure_noncurrent_retention",
"abort_upload",
"bootstrap", "epoch_acquire", "epoch_renew", "epoch_release", "epoch_enqueue",
"epoch_handoff",
];
const READS: &[&str] = &[
"head", "get_whole", "get_range", "get_range_segments", "list", "head_version",
"get_version", "list_versions", "presign_get", "lifecycle_rules", "list_uploads",
"epoch_read", "min_part_size", "max_parts", "upload_gate", "request_counts",
];
fn declared(src: &str, opener: &str) -> Vec<String> {
let lines: Vec<&str> = src.lines().collect();
let start = lines
.iter()
.position(|l| l.trim() == opener)
.unwrap_or_else(|| panic!("no line `{opener}` in the source"));
let indent = &lines[start][..lines[start].len() - lines[start].trim_start().len()];
let close = format!("{indent}}}");
let end = start + lines[start..].iter().position(|l| *l == close).expect("the block closes");
let mut names: Vec<String> = lines[start + 1..end]
.iter()
.filter_map(|l| {
let l = l.trim_start();
let rest = l.strip_prefix("async fn ").or_else(|| l.strip_prefix("fn "))?;
Some(rest[..rest.find(|c: char| !(c.is_alphanumeric() || c == '_'))?].to_string())
})
.collect();
names.sort();
names
}
#[test]
fn every_trait_method_is_filed_and_every_one_is_implemented_by_name() {
let mut filed: Vec<String> = WRITES.iter().chain(READS).map(|s| s.to_string()).collect();
filed.sort();
let mut dedup = filed.clone();
dedup.dedup();
assert_eq!(filed, dedup, "a method is filed twice");
let trait_fns = declared(include_str!("lib.rs"), "pub trait ObjectStore: Send + Sync {");
assert!(trait_fns.contains(&"put_whole".to_string()) && trait_fns.contains(&"request_counts".to_string()));
assert_eq!(trait_fns, filed, "the trait's methods and the read/write filing disagree");
let src = include_str!("readonly.rs");
assert_eq!(
declared(src, "impl<S: ObjectStore + ?Sized> ObjectStore for ReadOnly<S> {"),
filed,
"ReadOnly must name every trait method, defaulted ones included"
);
assert_eq!(
declared(src, "impl ObjectStore for Recorder {"),
filed,
"the recording double must name every trait method, or it cannot see the wrapper call one"
);
}
struct Recorder {
inner: MemoryStore,
calls: Mutex<Vec<&'static str>>,
}
impl Recorder {
fn saw(&self, m: &'static str) {
self.calls.lock().unwrap().push(m);
}
fn take(&self) -> Vec<&'static str> {
std::mem::take(&mut *self.calls.lock().unwrap())
}
}
#[async_trait]
impl ObjectStore for Recorder {
async fn put_whole(
&self,
key: &str,
body: Bytes,
condition: &PutCondition,
stamps: &GenerationStamps,
crc64: u64,
) -> StoreResult<ObjectMeta> {
self.saw("put_whole");
self.inner.put_whole(key, body, condition, stamps, crc64).await
}
async fn copy_object(
&self,
src_key: &str,
src_if_match: Option<&str>,
dst_key: &str,
condition: &PutCondition,
stamps: &GenerationStamps,
) -> StoreResult<ObjectMeta> {
self.saw("copy_object");
self.inner.copy_object(src_key, src_if_match, dst_key, condition, stamps).await
}
async fn compose_generation(&self, spec: &ComposeSpec<'_>) -> StoreResult<ObjectMeta> {
self.saw("compose_generation");
self.inner.compose_generation(spec).await
}
async fn head(&self, key: &str) -> StoreResult<ObjectMeta> {
self.saw("head");
self.inner.head(key).await
}
async fn get_whole(&self, key: &str, if_match: Option<&str>) -> StoreResult<(ObjectMeta, Bytes)> {
self.saw("get_whole");
self.inner.get_whole(key, if_match).await
}
async fn get_range(&self, key: &str, offset: u64, len: u64, if_match: &str) -> StoreResult<Bytes> {
self.saw("get_range");
self.inner.get_range(key, offset, len, if_match).await
}
async fn get_range_segments(
&self,
key: &str,
offset: u64,
len: u64,
if_match: &str,
) -> StoreResult<Vec<Bytes>> {
self.saw("get_range_segments");
self.inner.get_range_segments(key, offset, len, if_match).await
}
async fn list(&self, prefix: &str) -> StoreResult<Vec<ListedObject>> {
self.saw("list");
self.inner.list(prefix).await
}
async fn delete(&self, key: &str) -> StoreResult<()> {
self.saw("delete");
self.inner.delete(key).await
}
async fn delete_if_match(&self, key: &str, etag: &str) -> StoreResult<()> {
self.saw("delete_if_match");
self.inner.delete_if_match(key, etag).await
}
async fn delete_many(&self, keys: &[String]) -> StoreResult<DeleteManyReport> {
self.saw("delete_many");
self.inner.delete_many(keys).await
}
async fn head_version(&self, key: &str, version_id: &str) -> StoreResult<ObjectMeta> {
self.saw("head_version");
self.inner.head_version(key, version_id).await
}
async fn get_version(&self, key: &str, version_id: &str) -> StoreResult<(ObjectMeta, Bytes)> {
self.saw("get_version");
self.inner.get_version(key, version_id).await
}
async fn delete_version(&self, key: &str, version_id: &str) -> StoreResult<()> {
self.saw("delete_version");
self.inner.delete_version(key, version_id).await
}
async fn list_versions(&self, prefix: &str) -> StoreResult<Vec<ListedVersion>> {
self.saw("list_versions");
self.inner.list_versions(prefix).await
}
async fn presign_get(&self, key: &str, ttl_secs: u64) -> StoreResult<String> {
self.saw("presign_get");
self.inner.presign_get(key, ttl_secs).await
}
async fn presign_put(
&self,
key: &str,
ttl_secs: u64,
sha256: &[u8; 32],
) -> StoreResult<PresignedPut> {
self.saw("presign_put");
self.inner.presign_put(key, ttl_secs, sha256).await
}
async fn lifecycle_rules(&self) -> StoreResult<Vec<LifecycleView>> {
self.saw("lifecycle_rules");
self.inner.lifecycle_rules().await
}
async fn ensure_noncurrent_retention(&self, prefix: &str, days: u64) -> StoreResult<RetentionOutcome> {
self.saw("ensure_noncurrent_retention");
self.inner.ensure_noncurrent_retention(prefix, days).await
}
async fn list_uploads(&self, prefix: &str) -> StoreResult<Vec<PendingUpload>> {
self.saw("list_uploads");
self.inner.list_uploads(prefix).await
}
async fn abort_upload(&self, key: &str, upload_id: &str) -> StoreResult<()> {
self.saw("abort_upload");
self.inner.abort_upload(key, upload_id).await
}
async fn bootstrap(&self, prefix: &str) -> StoreResult<BootstrapReport> {
self.saw("bootstrap");
self.inner.bootstrap(prefix).await
}
async fn epoch_read(&self, key: &str) -> StoreResult<Option<EpochState>> {
self.saw("epoch_read");
self.inner.epoch_read(key).await
}
async fn epoch_acquire(
&self,
key: &str,
holder_id: &str,
supersede: Option<&EpochState>,
) -> StoreResult<EpochLease> {
self.saw("epoch_acquire");
self.inner.epoch_acquire(key, holder_id, supersede).await
}
async fn epoch_renew(&self, key: &str, lease: &EpochLease, echo: Option<&str>) -> StoreResult<EpochLease> {
self.saw("epoch_renew");
self.inner.epoch_renew(key, lease, echo).await
}
async fn epoch_release(&self, key: &str, lease: &EpochLease) -> StoreResult<()> {
self.saw("epoch_release");
self.inner.epoch_release(key, lease).await
}
async fn epoch_enqueue(&self, key: &str, observed: &EpochState, holder_id: &str) -> StoreResult<EpochState> {
self.saw("epoch_enqueue");
self.inner.epoch_enqueue(key, observed, holder_id).await
}
async fn epoch_handoff(&self, key: &str, lease: &EpochLease, echo: Option<&str>) -> StoreResult<()> {
self.saw("epoch_handoff");
self.inner.epoch_handoff(key, lease, echo).await
}
fn min_part_size(&self) -> u64 {
self.saw("min_part_size");
self.inner.min_part_size()
}
fn max_parts(&self) -> usize {
self.saw("max_parts");
self.inner.max_parts()
}
fn upload_gate(&self) -> Option<Arc<gate::ByteGate>> {
self.saw("upload_gate");
self.inner.upload_gate()
}
fn request_counts(&self) -> Option<RequestCounts> {
self.saw("request_counts");
self.inner.request_counts()
}
}
fn stamps() -> GenerationStamps {
GenerationStamps { generation: 1, epoch: 0, flush_uuid: "ro-test".into(), boundary_source: None, posix: None }
}
#[tokio::test]
async fn a_read_only_store_refuses_every_write_unsent_and_forwards_every_read() {
let mut mem = MemoryStore::new().with_upload_inflight_max_bytes(1 << 20);
mem.min_part = 7;
mem.max_parts = 9;
mem.plant_lifecycle_rule(LifecycleView {
id: "customer".into(),
enabled: true,
prefix: "p/".into(),
noncurrent_days: Some(3),
expired_delete_marker: false,
});
let rec = Arc::new(Recorder { inner: mem, calls: Mutex::new(vec![]) });
let body = Bytes::from_static(b"hello, read-only world");
let crc = crate::crc64_nvme(&body);
let v1 = rec.inner.put_whole("p/k", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(), crc).await.unwrap();
let v2 = rec
.inner
.put_whole("p/k", body.clone(), &PutCondition::IfMatch(v1.etag.clone()), &stamps(), crc)
.await
.unwrap();
let head_before = rec.inner.head("p/k").await.unwrap();
let lease = rec.inner.epoch_acquire("p/epoch", "holder", None).await.unwrap();
let cell = rec.inner.epoch_read("p/epoch").await.unwrap().unwrap();
let upload_id = rec.inner.raw_begin_upload("p/mpu");
rec.inner.reset_op_counts();
rec.take();
let ro = ReadOnly::new(rec.clone());
let mut refused: Vec<&str> = vec![];
let mut expect_refused = |verb: &'static str, r: StoreResult<()>| {
match r {
Err(StoreError::Auth(m)) if m.starts_with(&format!("read-only store: {verb} ")) => {}
other => panic!("{verb} on a read-only store answered {other:?}, not Auth"),
}
assert_eq!(rec.take(), Vec::<&str>::new(), "{verb} reached the inner store");
assert_eq!(rec.inner.total_ops(), 0, "{verb} sent a request: {:?}", rec.inner.op_counts());
refused.push(verb);
};
let spec = ComposeSpec {
key: "p/composed",
local: None,
parts: vec![PartSource::Local { offset: 0, len: 1 }],
base_key: None,
base_etag: None,
condition: PutCondition::Unconditional,
stamps: stamps(),
crc64: Some(0),
progress: None,
};
let cond = PutCondition::Unconditional;
expect_refused("put_whole", ro.put_whole("p/new", body.clone(), &cond, &stamps(), crc).await.map(drop));
expect_refused("copy_object", ro.copy_object("p/k", None, "p/copy", &cond, &stamps()).await.map(drop));
expect_refused("compose_generation", ro.compose_generation(&spec).await.map(drop));
expect_refused("delete", ro.delete("p/k").await);
expect_refused("delete_if_match", ro.delete_if_match("p/k", &v2.etag).await);
expect_refused("delete_many", ro.delete_many(&["p/k".to_string()]).await.map(drop));
expect_refused("delete_version", ro.delete_version("p/k", v1.version_id.as_deref().unwrap()).await);
expect_refused("presign_put", ro.presign_put("p/k", 60, &[0; 32]).await.map(drop));
expect_refused("ensure_noncurrent_retention", ro.ensure_noncurrent_retention("p/", 30).await.map(drop));
expect_refused("abort_upload", ro.abort_upload("p/mpu", &upload_id).await);
expect_refused("bootstrap", ro.bootstrap("p/").await.map(drop));
expect_refused("epoch_acquire", ro.epoch_acquire("p/epoch2", "h2", None).await.map(drop));
expect_refused("epoch_renew", ro.epoch_renew("p/epoch", &lease, None).await.map(drop));
expect_refused("epoch_release", ro.epoch_release("p/epoch", &lease).await);
expect_refused("epoch_enqueue", ro.epoch_enqueue("p/epoch", &cell, "h3").await.map(drop));
expect_refused("epoch_handoff", ro.epoch_handoff("p/epoch", &lease, None).await);
assert_eq!(refused, WRITES, "every write, and only the writes, were exercised");
assert_eq!(rec.inner.head("p/k").await.unwrap(), head_before);
assert_eq!(rec.inner.version_count("p/k"), 2);
assert_eq!(rec.inner.epoch_read("p/epoch").await.unwrap().unwrap(), cell);
assert_eq!(rec.inner.list_uploads("p/").await.unwrap().len(), 1);
rec.take();
let mut forwarded: Vec<&str> = vec![];
let mut reached = |verb: &'static str| {
assert_eq!(rec.take(), vec![verb], "{verb} did not reach the inner store as itself");
forwarded.push(verb);
};
let vid1 = v1.version_id.clone().unwrap();
assert_eq!(ro.head("p/k").await.unwrap(), head_before);
reached("head");
let whole = ro.get_whole("p/k", Some(&v2.etag)).await.unwrap();
assert_eq!(whole, rec.inner.get_whole("p/k", Some(&v2.etag)).await.unwrap());
assert_eq!(whole.1, body);
reached("get_whole");
assert_eq!(ro.get_range("p/k", 7, 9, &v2.etag).await.unwrap(), body.slice(7..16));
reached("get_range");
assert_eq!(ro.get_range_segments("p/k", 7, 9, &v2.etag).await.unwrap().concat(), body.slice(7..16));
reached("get_range_segments");
assert_eq!(ro.list("p/").await.unwrap(), rec.inner.list("p/").await.unwrap());
reached("list");
assert_eq!(ro.head_version("p/k", &vid1).await.unwrap(), rec.inner.head_version("p/k", &vid1).await.unwrap());
reached("head_version");
assert_eq!(ro.get_version("p/k", &vid1).await.unwrap(), rec.inner.get_version("p/k", &vid1).await.unwrap());
reached("get_version");
let versions = ro.list_versions("p/").await.unwrap();
assert_eq!(versions, rec.inner.list_versions("p/").await.unwrap());
reached("list_versions");
assert!(versions.iter().filter(|v| v.key == "p/k").count() == 2, "{versions:?}");
assert_eq!(ro.presign_get("p/k", 60).await.unwrap(), rec.inner.presign_get("p/k", 60).await.unwrap());
reached("presign_get");
let rules = ro.lifecycle_rules().await.unwrap();
assert_eq!(rules, rec.inner.lifecycle_rules().await.unwrap());
assert_eq!(rules.len(), 1);
reached("lifecycle_rules");
assert_eq!(ro.list_uploads("p/").await.unwrap(), rec.inner.list_uploads("p/").await.unwrap());
reached("list_uploads");
assert_eq!(ro.epoch_read("p/epoch").await.unwrap(), Some(cell.clone()));
reached("epoch_read");
assert_eq!(ro.min_part_size(), 7);
reached("min_part_size");
assert_eq!(ro.max_parts(), 9);
reached("max_parts");
let gate = ro.upload_gate().expect("the inner store has a byte gate");
assert!(Arc::ptr_eq(&gate, &rec.inner.upload_gate().unwrap()));
reached("upload_gate");
let counts = ro.request_counts().expect("the inner store counts");
assert_eq!(counts, rec.inner.request_counts().unwrap());
assert!(counts.get > 0 && counts.put == 0, "{counts:?}");
reached("request_counts");
assert_eq!(forwarded, READS, "every read, and only the reads, were exercised");
let writes: Vec<_> = rec
.inner
.op_counts()
.into_iter()
.filter(|(op, _)| !READS.contains(op))
.collect();
assert_eq!(writes, vec![], "a read sent a write");
}
}