coven_storage/
conformance.rs

1use crate::*;
2use coven_foundation::id_source::{DeviceId, ObjectId, StoreId};
3use std::num::NonZeroU64;
4use std::sync::Arc;
5
6/// Run identical object, range, prefix and immutable-create checks on any provider.
7/// The caller supplies the capability at construction, like production composition.
8pub struct Conformance {
9    storage: Arc<dyn Storage>,
10    store: StoreId,
11}
12/// The provider refused an occupied immutable path and left it unchanged.
13fn occupied<T>(result: Result<T, StorageError>) -> bool {
14    result.is_err_and(|error| error.own_failure() == Some(StorageFailure::AlreadyExists))
15}
16
17impl Conformance {
18    /// The provider under test, with an unused store ID.
19    pub fn new(storage: Arc<dyn Storage>, store: StoreId) -> Self {
20        Self { storage, store }
21    }
22    /// Two stores may publish the same device and number in one provider location.
23    pub async fn shared_location(&self, other: StoreId) {
24        assert_ne!(self.store, other);
25        let first = ObjectPath::slot(self.store, NonZeroU64::MIN);
26        let second = ObjectPath::slot(other, NonZeroU64::MIN);
27        let (left, right) = tokio::join!(
28            self.storage.create(&first, b"first"),
29            self.storage.create(&second, b"second")
30        );
31        left.unwrap();
32        right.unwrap();
33        for (store, path, bytes) in
34            [(self.store, &first, b"first".as_slice()), (other, &second, b"second".as_slice())]
35        {
36            assert_eq!(self.storage.read(path).await.unwrap(), bytes);
37            let listed = self.storage.list(&ObjectPrefix::all(store)).await.unwrap();
38            assert_eq!(listed.len(), 1);
39            assert_eq!(&listed[0].path, path);
40        }
41        self.storage.delete(&first).await.unwrap();
42        assert_eq!(self.storage.read(&second).await.unwrap(), b"second");
43        self.storage.delete(&second).await.unwrap();
44    }
45    /// Exercise the real provider operations, returning the first violated contract.
46    pub async fn run(&self) -> Result<(), StorageError> {
47        use coven_foundation::id_source::DeviceId;
48        let path = ObjectPath::file(self.store, DeviceId(31), ObjectId(uuid::Uuid::from_u128(1)));
49        let other = ObjectPath::file(self.store, DeviceId(32), ObjectId(uuid::Uuid::from_u128(1)));
50        let data = b"encrypted bytes";
51        if self.storage.status(&path).await?.is_some() {
52            return Err(StorageFailure::Protocol.with_source("absent exact status"));
53        }
54        let created = self.storage.create(&path, data).await?;
55        self.storage.create(&other, b"other").await?;
56        let first = self.storage.status(&path).await?.ok_or(StorageFailure::NotFound)?;
57        if created.is_some_and(|created| created != first) {
58            return Err(
59                StorageFailure::Protocol.with_source("create reply differs from exact status")
60            );
61        }
62        let read = self.storage.read_object(&path, None).await?;
63        if read.metadata != first {
64            return Err(
65                StorageFailure::Protocol.with_source("read metadata differs from exact status")
66            );
67        }
68        if read.bytes != data {
69            return Err(StorageFailure::Protocol.with_source("whole read"));
70        }
71        for (start, end) in [(0, 3), (4, 8), (12, 15)] {
72            let range = ByteRange::new(start, end)?;
73            let read = self.storage.read_object(&path, Some(range)).await?;
74            if read.metadata.size != first.size || read.metadata.revision != first.revision {
75                return Err(
76                    StorageFailure::Protocol.with_source("range omitted complete-object metadata")
77                );
78            }
79            if read.bytes != range.select(data)? {
80                return Err(StorageFailure::Protocol.with_source("range read"));
81            }
82        }
83        if !matches!(
84            self.storage.read_range(&path, ByteRange::new(15, 16)?).await,
85            Err(error) if error.failure() == StorageFailure::InvalidRange
86        ) {
87            return Err(StorageFailure::Protocol.with_source("past-end range misclassified"));
88        }
89        if !occupied(self.storage.create(&path, b"replacement").await) {
90            return Err(StorageFailure::Protocol.with_source("create-once refusal"));
91        }
92        if !occupied(self.storage.create(&path, data).await) {
93            return Err(StorageFailure::Protocol.with_source("create-once retry refusal"));
94        }
95        if self.storage.status(&path).await?.as_ref() != Some(&first) {
96            return Err(StorageFailure::Protocol
97                .with_source("immutable retry changed publication metadata"));
98        }
99        let listed = self
100            .storage
101            .list(&ObjectPrefix::device_files(self.store, DeviceId(31)))
102            .await?
103            .into_iter()
104            .map(|object| {
105                (object.path, object.size, object.stored_at, object.object_id, object.revision)
106            })
107            .collect::<Vec<_>>();
108        if listed
109            != [(
110                path.clone(),
111                first.size,
112                first.stored_at,
113                first.object_id.clone(),
114                first.revision.clone(),
115            )]
116        {
117            return Err(StorageFailure::Protocol.with_source(format!(
118                "prefix listing metadata: listed={listed:?}, status={first:?}"
119            )));
120        }
121        self.storage.delete(&path).await?;
122        self.storage.delete(&path).await?;
123        self.storage.delete(&other).await?;
124        if self.storage.status(&path).await?.is_some() {
125            return Err(StorageFailure::Protocol.with_source("deleted exact status"));
126        }
127        if self.storage.read(&path).await.err().map(|e| e.failure())
128            != Some(StorageFailure::NotFound)
129        {
130            return Err(StorageFailure::Protocol.with_source("deleted object read"));
131        }
132        if !matches!(
133            self.storage.replace(&first, data).await,
134            Err(error) if error.failure() == StorageFailure::InvalidPath
135        ) {
136            return Err(
137                StorageFailure::Protocol.with_source("immutable object accepted replacement")
138            );
139        }
140        for path in [ObjectPath::floor(self.store)] {
141            self.storage.create(&path, b"last record").await?;
142            let mut previous = self.storage.status(&path).await?.ok_or(StorageFailure::NotFound)?;
143            for bytes in [b"next record", b"last record"] {
144                self.storage.replace(&previous, bytes).await?;
145                let next = self.storage.status(&path).await?.ok_or(StorageFailure::NotFound)?;
146                if previous.revision == next.revision {
147                    return Err(
148                        StorageFailure::Protocol.with_source("replacement did not change revision")
149                    );
150                }
151                if !matches!(self.storage.replace(&previous, b"stale floor").await, Err(error) if error.failure() == StorageFailure::Conflict)
152                {
153                    return Err(StorageFailure::Protocol
154                        .with_source("stale floor replacement was accepted"));
155                }
156                previous = next;
157            }
158            if self.storage.read(&path).await? != b"last record" {
159                return Err(StorageFailure::Protocol.with_source("object replacement"));
160            }
161            if !self.storage.list(&ObjectPrefix::all(self.store)).await?.iter().any(|object| {
162                object.path == path
163                    && object.size == previous.size
164                    && object.revision == previous.revision
165            }) {
166                return Err(StorageFailure::Protocol
167                    .with_source("replaced object metadata differs from listing"));
168            }
169            self.storage.delete(&path).await?;
170        }
171        self.storage.create(&path, &[]).await?;
172        if !occupied(self.storage.create(&path, &[]).await) {
173            return Err(StorageFailure::Protocol.with_source("empty create-once refusal"));
174        }
175        if !self.storage.read(&path).await?.is_empty() {
176            return Err(StorageFailure::Protocol.with_source("empty object changed"));
177        }
178        if !occupied(self.storage.create(&path, b"different").await) {
179            return Err(StorageFailure::Protocol.with_source("empty create-once refusal"));
180        }
181        if !self.storage.read(&path).await?.is_empty() {
182            return Err(StorageFailure::Protocol.with_source("empty immutable object replaced"));
183        }
184        self.storage.delete(&path).await?;
185        let bytes: Vec<_> = (0..42).map(|index| (index % 251) as u8).collect();
186        self.storage.create(&path, &bytes).await?;
187        if self.storage.read_range(&path, ByteRange::new(5, 37)?).await? != bytes[5..37] {
188            return Err(StorageFailure::Protocol.with_source("range across upload parts changed"));
189        }
190        if self
191            .storage
192            .read_range(&path, ByteRange::new(bytes.len() as u64 - 1, bytes.len() as u64 + 1)?)
193            .await
194            .err()
195            .map(|error| error.failure())
196            != Some(StorageFailure::InvalidRange)
197        {
198            return Err(
199                StorageFailure::Protocol.with_source("clipped past-end range misclassified")
200            );
201        }
202        self.storage.delete(&path).await?;
203        Ok(())
204    }
205
206    /// Setups of one store race for its first sequence slot (§10.3, D10): of three
207    /// concurrent creates, one per claimant connection, exactly one wins and
208    /// keeps its bytes, and the others are refused as occupied. The winner's
209    /// connection deletes the slot afterwards.
210    pub async fn creation_race(
211        &self,
212        claimants: [Arc<dyn Storage>; 3],
213    ) -> Result<(), StorageError> {
214        let marker = ObjectPath::slot(self.store, NonZeroU64::MIN);
215        let claims: [&[u8]; 3] = [b"claim 1", b"claim 2", b"claim 3"];
216        let (first, second, third) = tokio::join!(
217            claimants[0].create(&marker, claims[0]),
218            claimants[1].create(&marker, claims[1]),
219            claimants[2].create(&marker, claims[2]),
220        );
221        let mut winners = Vec::new();
222        for (index, result) in [first, second, third].into_iter().enumerate() {
223            match result {
224                Ok(_) => winners.push(index),
225                Err(error) if error.own_failure() == Some(StorageFailure::AlreadyExists) => {}
226                Err(error) => return Err(error),
227            }
228        }
229        let [winner] = winners.as_slice() else {
230            return Err(StorageFailure::Protocol
231                .with_source(format!("{} concurrent creates won one path", winners.len())));
232        };
233        if self.storage.read(&marker).await? != claims[*winner] {
234            return Err(StorageFailure::Protocol.with_source("creation race kept another claim"));
235        }
236        claimants[*winner].delete(&marker).await
237    }
238}
239
240impl Conformance {
241    /// Preserve provider publication times and sizes through listings and retries.
242    /// Advance the fixture clock after publication to distinguish retry timestamps.
243    pub async fn listing_times(
244        &self,
245        stored_at: coven_format::value::StorageTime,
246        advance: impl std::future::Future<Output = ()>,
247    ) {
248        let storage = &self.storage;
249        let first = ObjectPath::file(self.store, DeviceId(31), ObjectId(uuid::Uuid::from_u128(1)));
250        let second = ObjectPath::file(self.store, DeviceId(32), ObjectId(uuid::Uuid::from_u128(1)));
251        storage.create(&first, b"first").await.unwrap();
252        storage.create(&second, b"second").await.unwrap();
253        let expected = storage.list(&ObjectPrefix::files(self.store)).await.unwrap();
254        assert_eq!(
255            expected
256                .iter()
257                .map(|object| (&object.path, object.size, object.stored_at))
258                .collect::<Vec<_>>(),
259            [(&first, 5, stored_at), (&second, 6, stored_at)]
260        );
261        advance.await;
262        assert!(occupied(storage.create(&first, b"first").await));
263        assert_eq!(storage.list(&ObjectPrefix::all(self.store)).await.unwrap(), expected);
264        storage.delete(&first).await.unwrap();
265        storage.delete(&second).await.unwrap();
266    }
267
268    /// Restart an expired recording at its original destination with no progress.
269    /// Return the expiry error so native fixtures can also inspect its provider cause.
270    pub async fn expired_session(
271        &self,
272        expire: impl std::future::Future<Output = ()>,
273    ) -> StorageError {
274        let storage = &self.storage;
275        let path = ObjectPath::file(self.store, DeviceId(31), ObjectId(uuid::Uuid::from_u128(1)));
276        let mut upload = storage.begin_upload(&path, 4).await.unwrap();
277        storage.resume_upload(&mut upload).await.unwrap();
278        expire.await;
279        let error = storage.resume_upload(&mut upload).await.unwrap_err();
280        assert_eq!(error.failure(), StorageFailure::SessionExpired);
281        let replacement = storage.restart_upload(&upload).await.unwrap();
282        assert_eq!(replacement.path(), &path);
283        assert_eq!(replacement.total_bytes(), 4);
284        assert_eq!(replacement.confirmed_bytes(), 0);
285        assert_ne!(replacement.encode().unwrap().as_bytes(), upload.encode().unwrap().as_bytes());
286        let mut replacement =
287            UploadSession::decode(replacement.encode().unwrap().as_bytes()).unwrap();
288        storage.resume_upload(&mut replacement).await.unwrap();
289        storage.upload_part(&mut replacement, b"data").await.unwrap();
290        storage.finish_upload(&mut replacement).await.unwrap();
291        assert_eq!(storage.read(&path).await.unwrap(), b"data");
292        assert!(matches!(storage.restart_upload(&replacement).await,
293            Err(error) if error.failure() == StorageFailure::InvalidPart));
294        storage.delete(&path).await.unwrap();
295        error
296    }
297
298    /// Non-owner accounts may neither add nor remove another account's access.
299    pub async fn owner_only_sharing(&self) {
300        for error in [
301            self.storage.grant_access("[email protected]").await.err().unwrap(),
302            self.storage.revoke_access("[email protected]").await.err().unwrap(),
303        ] {
304            assert_eq!(error.failure(), StorageFailure::NotStoreOwner);
305        }
306    }
307
308    /// Restore a persisted recording after the provider accepts a part but loses its reply.
309    pub async fn lost_part_reply(
310        &self,
311        part_size: usize,
312        lose: impl std::future::Future<Output = ()>,
313    ) {
314        let storage = &self.storage;
315        let path = ObjectPath::file(self.store, DeviceId(31), ObjectId(uuid::Uuid::from_u128(1)));
316        let mut upload = storage.begin_upload(&path, 2 * part_size as u64 + 1).await.unwrap();
317        assert_eq!(upload.part_size(), part_size);
318        storage.resume_upload(&mut upload).await.unwrap();
319        // Where beginning sends nothing, the first part starts the session,
320        // and only the recording after it names the session (Appendix E.10).
321        let first = vec![41; part_size];
322        storage.upload_part(&mut upload, &first).await.unwrap();
323        let recorded = upload.encode().unwrap();
324        let bytes = vec![42; part_size];
325        lose.await;
326        assert_eq!(
327            storage.upload_part(&mut upload, &bytes).await.unwrap_err().failure(),
328            StorageFailure::Network
329        );
330        assert_eq!(upload.confirmed_bytes(), part_size as u64);
331        let mut upload = UploadSession::decode(recorded.as_bytes()).unwrap();
332        storage.resume_upload(&mut upload).await.unwrap();
333        assert_eq!(upload.confirmed_bytes(), 2 * part_size as u64);
334        storage.upload_part(&mut upload, b"z").await.unwrap();
335        storage.finish_upload(&mut upload).await.unwrap();
336        storage.finish_upload(&mut upload).await.unwrap();
337        let mut expected = first;
338        expected.extend(bytes);
339        expected.push(b'z');
340        assert_eq!(storage.read(&path).await.unwrap(), expected);
341        storage.delete(&path).await.unwrap();
342    }
343
344    /// Every object, upload and sharing operation preserves a permission failure.
345    pub async fn permission_failures(
346        &self,
347        deny: impl std::future::Future<Output = ()>,
348    ) -> Vec<StorageError> {
349        let storage = &self.storage;
350        let path = ObjectPath::slot(self.store, NonZeroU64::MIN);
351        let floor = ObjectPath::floor(self.store);
352        storage.create(&floor, b"previous floor").await.unwrap();
353        let previous = storage.status(&floor).await.unwrap().unwrap();
354        let mut ready = storage.begin_upload(&path, 4).await.unwrap();
355        let part = vec![1; ready.part_size()];
356        let mut pending = storage.begin_upload(&path, part.len() as u64 + 1).await.unwrap();
357        storage.upload_part(&mut pending, &part).await.unwrap();
358        storage.upload_part(&mut ready, b"data").await.unwrap();
359        deny.await;
360        let mut errors = vec![
361            storage.create(&path, b"data").await.err().unwrap(),
362            storage.replace(&previous, b"data").await.err().unwrap(),
363            storage.read(&path).await.err().unwrap(),
364            storage.status(&path).await.err().unwrap(),
365            storage.read_range(&path, ByteRange::new(0, 1).unwrap()).await.err().unwrap(),
366            storage.list(&ObjectPrefix::all(self.store)).await.err().unwrap(),
367            storage.delete(&path).await.err().unwrap(),
368            storage.resume_upload(&mut pending).await.err().unwrap(),
369            storage.upload_part(&mut pending, b"z").await.err().unwrap(),
370            storage.finish_upload(&mut ready).await.err().unwrap(),
371            storage.abort_upload(&pending).await.err().unwrap(),
372        ];
373        if !matches!(storage.config().provider(), CloudProvider::S3 | CloudProvider::OneDrive) {
374            errors.push(storage.begin_upload(&path, 4).await.err().unwrap());
375        }
376        if storage.config().provider() != CloudProvider::S3 {
377            errors.push(storage.grant_access("member").await.err().unwrap());
378            errors.push(storage.revoke_access("member").await.err().unwrap());
379        }
380        for error in &errors {
381            assert_eq!(error.failure(), StorageFailure::PermissionDenied, "{error:?}");
382        }
383        errors
384    }
385}
386
387impl Conformance {
388    /// Competing recorded floor uploads keep the original absent/revision
389    /// condition through parts, serialization, resumption and publication.
390    pub async fn conditional_floor_upload(&self) {
391        let storage = &self.storage;
392        let path = ObjectPath::floor(self.store);
393        for seeded in [false, true] {
394            if seeded {
395                storage.create(&path, b"previous").await.unwrap();
396            }
397            let previous = storage.status(&path).await.unwrap();
398            // Ask the provider for its part boundary before constructing the
399            // contending transfers; aborting this empty session publishes nothing.
400            let probe = match &previous {
401                Some(previous) => storage.begin_replace(previous, 1).await.unwrap(),
402                None => storage.begin_upload(&path, 1).await.unwrap(),
403            };
404            let part_size = probe.part_size();
405            storage.abort_upload(&probe).await.unwrap();
406            let total = part_size as u64 + 1;
407            let mut uploads = Vec::new();
408            for _ in 0..2 {
409                uploads.push(match &previous {
410                    Some(previous) => storage.begin_replace(previous, total).await.unwrap(),
411                    None => storage.begin_upload(&path, total).await.unwrap(),
412                });
413            }
414            for (index, upload) in uploads.iter_mut().enumerate() {
415                storage.upload_part(upload, &vec![index as u8; part_size]).await.unwrap();
416                assert_eq!(storage.status(&path).await.unwrap(), previous);
417                let target = upload.target().clone();
418                *upload = UploadSession::decode(upload.encode().unwrap().as_bytes()).unwrap();
419                storage.resume_upload(upload).await.unwrap();
420                assert_eq!(upload.confirmed_bytes(), part_size as u64);
421                assert_eq!(upload.target(), &target);
422            }
423            storage.upload_part(&mut uploads[0], b"a").await.unwrap();
424            storage.finish_upload(&mut uploads[0]).await.unwrap();
425            let result = match storage.upload_part(&mut uploads[1], b"b").await {
426                Ok(()) => storage.finish_upload(&mut uploads[1]).await,
427                Err(error) => Err(error),
428            };
429            let error = result.unwrap_err();
430            assert_eq!(
431                error.own_failure(),
432                Some(if seeded { StorageFailure::Conflict } else { StorageFailure::AlreadyExists })
433            );
434            for upload in &uploads {
435                storage.abort_upload(upload).await.unwrap();
436            }
437            let mut expected = vec![0; part_size];
438            expected.push(b'a');
439            assert_eq!(storage.read(&path).await.unwrap(), expected);
440            let current = storage.status(&path).await.unwrap().unwrap();
441            if let Some(previous) = previous {
442                assert_ne!(current.revision, previous.revision);
443            }
444            // The next buffered replacement must also retire a multipart floor.
445            storage.replace(&current, b"last").await.unwrap();
446            assert_eq!(storage.read(&path).await.unwrap(), b"last");
447            storage.delete(&path).await.unwrap();
448        }
449    }
450}
coven_storage/conformance.rs

coven_storage/
conformance.rs
