persistant_blob/
conformance.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Contract tests for the blob leaves whose written contract sequential calls can decide.
3//!
4//! [`certify`] runs each leaf's test on the target, on a correct reference backend (the positive
5//! control: the test can pass) and on reference backends with a [`Defect`] (the known-bad control:
6//! the test can fail). A leaf is recorded **on** only when the target passed and both controls
7//! behaved. Anything else is an off row with the reason.
8
9use crate::concurrency::{Task, join_all, yield_now};
10use crate::{BlobBackend, BlobError, Defect, MemoryBlob};
11use persistant_core::{CapabilityId, Config, Controls, Reason, Row, Verdict};
12use std::collections::BTreeMap;
13
14/// A leaf this module can decide.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16#[non_exhaustive]
17pub enum Leaf {
18    /// `blob.read`
19    Read,
20    /// `blob.write`
21    Write,
22    /// `blob.delete`
23    Delete,
24    /// `blob.stat`
25    Stat,
26    /// `blob.list`
27    List,
28    /// `blob.create_exclusive`, by sampled concurrency
29    CreateExclusive,
30    /// `blob.replace_atomic.concurrent_read`, by sampled concurrency
31    ReplaceConcurrentRead,
32    /// `blob.append`
33    Append,
34    /// `blob.append.concurrent_whole`, by sampled concurrency
35    AppendConcurrentWhole,
36}
37
38impl Leaf {
39    /// Every leaf with a test here.
40    pub const ALL: [Leaf; 9] = [
41        Leaf::Read,
42        Leaf::Write,
43        Leaf::Delete,
44        Leaf::Stat,
45        Leaf::List,
46        Leaf::CreateExclusive,
47        Leaf::ReplaceConcurrentRead,
48        Leaf::Append,
49        Leaf::AppendConcurrentWhole,
50    ];
51
52    /// The capability id.
53    pub fn id(self) -> &'static str {
54        match self {
55            Leaf::Read => "blob.read",
56            Leaf::Write => "blob.write",
57            Leaf::Delete => "blob.delete",
58            Leaf::Stat => "blob.stat",
59            Leaf::List => "blob.list",
60            Leaf::CreateExclusive => "blob.create_exclusive",
61            Leaf::ReplaceConcurrentRead => "blob.replace_atomic.concurrent_read",
62            Leaf::Append => "blob.append",
63            Leaf::AppendConcurrentWhole => "blob.append.concurrent_whole",
64        }
65    }
66
67    /// The trial parameters of a sampled-concurrency leaf, recorded in its row (a race that did
68    /// not occur in N trials is "no failure seen", and N goes in the row).
69    fn trials(self) -> &'static str {
70        match self {
71            Leaf::CreateExclusive => "contenders=16;rounds=20;value_bytes=4096",
72            Leaf::ReplaceConcurrentRead => "readers=4;replaces=40;value_bytes=65536",
73            Leaf::AppendConcurrentWhole => "contenders=16;records=10;value_bytes=64",
74            _ => "",
75        }
76    }
77
78    fn defects(self) -> &'static [Defect] {
79        match self {
80            Leaf::Read => &[Defect::ReadTruncates, Defect::KeysFoldCase],
81            Leaf::Write => &[Defect::WriteIsLost],
82            Leaf::Delete => &[Defect::DeleteIsIgnored],
83            Leaf::Stat => &[Defect::StatLiesAboutLength],
84            Leaf::List => &[
85                Defect::ListIsDescending,
86                Defect::ListPrefixIsPattern,
87                Defect::ListIgnoresPaging,
88            ],
89            Leaf::CreateExclusive => &[Defect::CreateNotExclusive],
90            Leaf::ReplaceConcurrentRead => &[Defect::ReplaceTearsReads],
91            Leaf::Append => &[Defect::AppendOverwrites],
92            Leaf::AppendConcurrentWhole => &[Defect::AppendTearsRecords, Defect::AppendOverwrites],
93        }
94    }
95}
96
97/// Why a leaf's test did not pass.
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub enum Failure {
100    /// The backend returned something other than the contract says.
101    Wrong(String),
102    /// The backend could not perform an operation the test needs.
103    Backend(BlobError),
104}
105
106impl From<BlobError> for Failure {
107    fn from(e: BlobError) -> Self {
108        Failure::Backend(e)
109    }
110}
111
112fn wrong<T>(msg: impl Into<String>) -> Result<T, Failure> {
113    Err(Failure::Wrong(msg.into()))
114}
115
116/// Run one leaf's test against `b`. Writes only under `persistant-conformance/<leaf>/`.
117pub async fn run_leaf<B: BlobBackend>(leaf: Leaf, b: &B) -> Result<(), Failure> {
118    let base = format!("persistant-conformance/{}/", leaf.id());
119    let result = match leaf {
120        Leaf::Read => read(b, &base).await,
121        Leaf::Write => write(b, &base).await,
122        Leaf::Delete => delete(b, &base).await,
123        Leaf::Stat => stat(b, &base).await,
124        Leaf::List => list(b, &base).await,
125        Leaf::CreateExclusive => create_exclusive(b, &base).await,
126        Leaf::ReplaceConcurrentRead => replace_concurrent_read(b, &base).await,
127        Leaf::Append => append(b, &base).await,
128        Leaf::AppendConcurrentWhole => append_concurrent_whole(b, &base).await,
129    };
130    // Best-effort cleanup so a shared target is left as found.
131    if let Ok(page) = b.list(&base, None, usize::MAX).await {
132        for e in page.entries {
133            let _ = b.delete(&e.key).await;
134        }
135    }
136    result
137}
138
139async fn read<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
140    let k = format!("{base}k");
141    if b.read(&k).await?.is_some() {
142        return wrong("an absent key read as present");
143    }
144    for value in [vec![], vec![0u8], vec![1, 2, 3, 0xFF, 0x00, 0x80]] {
145        b.write(&k, value.clone()).await?;
146        match b.read(&k).await? {
147            Some(got) if got == value => {}
148            Some(got) => return wrong(format!("wrote {value:?}, read {got:?}")),
149            None => return wrong(format!("wrote {value:?}, read nothing")),
150        }
151    }
152    // Keys that differ only in letter case are different keys. A case-insensitive filesystem
153    // (the Windows and default macOS ones) fails this.
154    let upper = format!("{base}CaseKey");
155    let lower = format!("{base}casekey");
156    b.write(&upper, b"U".to_vec()).await?;
157    b.write(&lower, b"l".to_vec()).await?;
158    match b.read(&upper).await? {
159        Some(v) if v == b"U" => {}
160        other => {
161            return wrong(format!(
162                "keys differing only in case are the same key (read {other:?} after writing both)"
163            ));
164        }
165    }
166    Ok(())
167}
168
169async fn write<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
170    let k = format!("{base}nested/dir/k");
171    b.write(&k, b"first".to_vec()).await?;
172    b.write(&k, b"second, longer".to_vec()).await?;
173    match b.read(&k).await? {
174        Some(v) if v == b"second, longer" => {}
175        other => return wrong(format!("after two writes read {other:?}")),
176    }
177    b.write(&k, b"x".to_vec()).await?;
178    match b.read(&k).await? {
179        Some(v) if v == b"x" => Ok(()),
180        other => wrong(format!("a shorter replacement left {other:?}")),
181    }
182}
183
184async fn delete<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
185    let k = format!("{base}k");
186    b.write(&k, b"v".to_vec()).await?;
187    b.delete(&k).await?;
188    if b.read(&k).await?.is_some() {
189        return wrong("a deleted key still reads");
190    }
191    b.delete(&k).await?; // deleting an absent key is not an error
192    Ok(())
193}
194
195async fn stat<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
196    let k = format!("{base}k");
197    if b.stat(&k).await?.is_some() {
198        return wrong("an absent key has a stat");
199    }
200    for len in [0usize, 1, 5, 4096] {
201        b.write(&k, vec![7u8; len]).await?;
202        match b.stat(&k).await? {
203            Some(e) if e.key == k && e.len == len as u64 => {}
204            other => return wrong(format!("a {len}-byte value stat as {other:?}")),
205        }
206    }
207    Ok(())
208}
209
210async fn list<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
211    // "1" < "10" < "2" bytewise: a numeric or natural sort would put "2" before "10".
212    let under = ["a/1", "a/10", "a/2", "a/b/c"];
213    for k in under
214        .iter()
215        .chain(["ab", "abz", "a0x", "a%z", "a[0]x", "b/1"].iter())
216    {
217        b.write(&format!("{base}{k}"), k.as_bytes().to_vec())
218            .await?;
219    }
220    let full = |s: &str| format!("{base}{s}");
221    let want: Vec<String> = ["a/1", "a/10", "a/2", "a/b/c"]
222        .iter()
223        .map(|s| full(s))
224        .collect();
225
226    // One page: literal prefix, ascending bytewise, complete.
227    let all = b.list(&full("a/"), None, 100).await?;
228    let got: Vec<String> = all.entries.iter().map(|e| e.key.clone()).collect();
229    if got != want || all.next.is_some() {
230        return wrong(format!(
231            "prefix a/ listed {got:?} next {:?}, want {want:?}",
232            all.next
233        ));
234    }
235    if all.entries.iter().any(|e| e.len == 0) {
236        return wrong("a listed entry reports length 0 for a non-empty value");
237    }
238
239    // The prefix is literal: `%` (SQL LIKE) and `[` (glob class) match only themselves. These
240    // are valid in a Windows file name; `*` and `?` are not, so they cannot be used here.
241    for (prefix, only) in [("a%", "a%z"), ("a[0]", "a[0]x")] {
242        let p = b.list(&full(prefix), None, 100).await?;
243        let got: Vec<String> = p.entries.iter().map(|e| e.key.clone()).collect();
244        if got != vec![full(only)] {
245            return wrong(format!(
246                "prefix {prefix} listed {got:?}, want only {:?}",
247                full(only)
248            ));
249        }
250    }
251
252    // Paging: limit 2 over 4 keys, each key once, continuation ends with None.
253    let mut seen = Vec::new();
254    let mut after: Option<String> = None;
255    for _ in 0..10 {
256        let p = b.list(&full("a/"), after.as_deref(), 2).await?;
257        if p.entries.len() > 2 {
258            return wrong(format!(
259                "a page of limit 2 held {} entries",
260                p.entries.len()
261            ));
262        }
263        seen.extend(p.entries.iter().map(|e| e.key.clone()));
264        match p.next {
265            Some(c) => after = Some(c),
266            None => break,
267        }
268    }
269    if seen != want {
270        return wrong(format!("paged listing returned {seen:?}, want {want:?}"));
271    }
272    Ok(())
273}
274
275async fn create_exclusive<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
276    const CONTENDERS: usize = 16;
277    const ROUNDS: usize = 20;
278    let payload = |i: usize| vec![i as u8; 4096];
279    for round in 0..ROUNDS {
280        let key = format!("{base}r{round}");
281        let key = key.as_str();
282        let tasks: Vec<Task<'_, Result<bool, BlobError>>> = (0..CONTENDERS)
283            .map(|i| -> Task<'_, _> { Box::pin(b.create_exclusive(key, payload(i))) })
284            .collect();
285        let results = join_all(tasks).await;
286        let mut winners = Vec::new();
287        for (i, r) in results.into_iter().enumerate() {
288            if r? {
289                winners.push(i);
290            }
291        }
292        if winners.len() != 1 {
293            return wrong(format!(
294                "round {round}: {} of {CONTENDERS} concurrent creators succeeded, want exactly 1",
295                winners.len()
296            ));
297        }
298        match b.read(key).await? {
299            Some(v) if v == payload(winners[0]) => {}
300            other => {
301                return wrong(format!(
302                    "round {round}: the stored value is not the winner's ({} bytes)",
303                    other.map_or(0, |v| v.len())
304                ));
305            }
306        }
307    }
308    Ok(())
309}
310
311async fn replace_concurrent_read<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
312    const LEN: usize = 65536;
313    const READERS: usize = 4;
314    const REPLACES: usize = 40;
315    let key = format!("{base}k");
316    let key = key.as_str();
317    let a = vec![b'A'; LEN];
318    let c = vec![b'B'; LEN];
319    b.replace_atomic(key, a.clone()).await?;
320    let done = std::cell::Cell::new(false);
321    let done = &done;
322    let mut tasks: Vec<Task<'_, Result<(), Failure>>> = Vec::new();
323    tasks.push(Box::pin(async move {
324        for i in 0..REPLACES {
325            let v = if i % 2 == 0 { c.clone() } else { a.clone() };
326            if let Err(e) = b.replace_atomic(key, v).await {
327                done.set(true);
328                return Err(Failure::from(e));
329            }
330            yield_now().await;
331        }
332        done.set(true);
333        Ok(())
334    }));
335    for _ in 0..READERS {
336        tasks.push(Box::pin(async move {
337            for _ in 0..(REPLACES * 20) {
338                match b.read(key).await {
339                    Ok(Some(v)) => {
340                        let uniform = v.len() == LEN
341                            && v.iter().all(|x| *x == v[0])
342                            && (v[0] == b'A' || v[0] == b'B');
343                        if !uniform {
344                            done.set(true);
345                            return wrong(format!(
346                                "a reader saw a mix of old and new ({} bytes, first {:?})",
347                                v.len(),
348                                v.first()
349                            ));
350                        }
351                    }
352                    Ok(None) => {
353                        done.set(true);
354                        return wrong("the key vanished during a replace");
355                    }
356                    Err(e) => {
357                        done.set(true);
358                        return Err(Failure::from(e));
359                    }
360                }
361                if done.get() {
362                    break;
363                }
364                yield_now().await;
365            }
366            Ok(())
367        }));
368    }
369    for r in join_all(tasks).await {
370        r?;
371    }
372    Ok(())
373}
374
375async fn append<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
376    let k = format!("{base}k");
377    b.write(&k, b"head".to_vec()).await?;
378    for piece in [&b"a"[..], b"bc", b"", b"def"] {
379        b.append(&k, piece.to_vec()).await?;
380    }
381    match b.read(&k).await? {
382        Some(v) if v == b"headabcdef" => Ok(()),
383        other => wrong(format!(
384            "appends in order left {other:?}, want \"headabcdef\""
385        )),
386    }
387}
388
389async fn append_concurrent_whole<B: BlobBackend>(b: &B, base: &str) -> Result<(), Failure> {
390    const CONTENDERS: usize = 16;
391    const RECORDS: usize = 10;
392    const LEN: usize = 64;
393    let key = format!("{base}k");
394    let key = key.as_str();
395    b.write(key, Vec::new()).await?;
396    let tasks: Vec<Task<'_, Result<(), BlobError>>> = (0..CONTENDERS)
397        .map(|i| -> Task<'_, _> {
398            Box::pin(async move {
399                for _ in 0..RECORDS {
400                    b.append(key, vec![i as u8; LEN]).await?;
401                    yield_now().await;
402                }
403                Ok(())
404            })
405        })
406        .collect();
407    for r in join_all(tasks).await {
408        r?;
409    }
410    let all = b.read(key).await?.unwrap_or_default();
411    if all.len() != CONTENDERS * RECORDS * LEN {
412        return wrong(format!(
413            "{} bytes after {} appends of {LEN}, want {}",
414            all.len(),
415            CONTENDERS * RECORDS,
416            CONTENDERS * RECORDS * LEN
417        ));
418    }
419    let mut counts = [0usize; CONTENDERS];
420    for chunk in all.chunks(LEN) {
421        let first = chunk[0] as usize;
422        if first >= CONTENDERS || chunk.iter().any(|x| *x as usize != first) {
423            return wrong("a record is interleaved with another's bytes");
424        }
425        counts[first] += 1;
426    }
427    if counts.iter().any(|c| *c != RECORDS) {
428        return wrong(format!(
429            "records per appender {counts:?}, want {RECORDS} each"
430        ));
431    }
432    Ok(())
433}
434
435/// The outcome of certifying one leaf.
436#[derive(Debug, Clone, PartialEq, Eq)]
437pub struct Certified {
438    /// The leaf.
439    pub leaf: Leaf,
440    /// The row to record.
441    pub row: Row,
442}
443
444/// Run every leaf here against `target` and return one row per leaf.
445///
446/// `config` is the facts the gate can read for this target (`collation=...`); `metadata` is
447/// copied into every row, with `test=blob::<leaf>` added. A leaf is **on** only if the target
448/// passed *and* the positive control passed *and* every known-bad control failed.
449pub async fn certify<B: BlobBackend>(
450    target: &B,
451    config: &Config,
452    metadata: &Config,
453) -> Vec<Certified> {
454    certify_leaves(target, config, metadata, &Leaf::ALL).await
455}
456
457/// As [`certify`], for the listed leaves only (each with its controls).
458pub async fn certify_leaves<B: BlobBackend>(
459    target: &B,
460    config: &Config,
461    metadata: &Config,
462    leaves: &[Leaf],
463) -> Vec<Certified> {
464    let mut out = Vec::new();
465    for leaf in leaves.iter().copied() {
466        let positive_passed = run_leaf(leaf, &MemoryBlob::new()).await.is_ok();
467        let mut known_bad_failed = true;
468        for d in leaf.defects() {
469            if run_leaf(leaf, &MemoryBlob::with_defect(*d)).await.is_ok() {
470                known_bad_failed = false;
471            }
472        }
473        let controls = Controls {
474            positive_passed,
475            known_bad_failed,
476        };
477        let result = run_leaf(leaf, target).await;
478        let mut md = metadata
479            .clone()
480            .with("test", format!("blob::{}", leaf.id()));
481        if let Ok(t) = persistant_core::Config::parse(leaf.trials()) {
482            for key in [
483                "contenders",
484                "rounds",
485                "value_bytes",
486                "readers",
487                "replaces",
488                "records",
489            ] {
490                if let Some(v) = t.get(key) {
491                    md = md.with(key, v);
492                }
493            }
494        }
495        let id = CapabilityId::new(leaf.id()).expect("leaf id is valid");
496        let row = match (
497            &result,
498            controls.positive_passed && controls.known_bad_failed,
499        ) {
500            (Ok(()), true) => Verdict::Exact
501                .into_row(
502                    id,
503                    target.backend(),
504                    target.version(),
505                    config.clone(),
506                    md,
507                    controls,
508                )
509                .expect("both controls hold"),
510            (Ok(()), false) => off(
511                id,
512                target,
513                config,
514                md,
515                Reason::Untested,
516                format!("controls did not hold: {controls:?}"),
517            ),
518            (Err(Failure::Wrong(m)), _) => off(id, target, config, md, Reason::Wrong, m.clone()),
519            (Err(Failure::Backend(e)), _) => {
520                off(id, target, config, md, Reason::Missing, e.to_string())
521            }
522        };
523        out.push(Certified { leaf, row });
524    }
525    out
526}
527
528fn off<B: BlobBackend>(
529    subject: CapabilityId,
530    target: &B,
531    config: &Config,
532    metadata: Config,
533    reason: Reason,
534    note: String,
535) -> Row {
536    Row {
537        subject_kind: "capability".to_string(),
538        subject,
539        backend: target.backend().to_string(),
540        version: target.version().to_string(),
541        config: config.clone(),
542        enabled: false,
543        reason: Some(reason),
544        metadata,
545        note: note.replace(['\t', '\n'], " "),
546        extra: BTreeMap::new(),
547    }
548}
549
550#[cfg(test)]
551mod tests {
552    use super::*;
553
554    #[tokio::test]
555    async fn a_correct_backend_is_on_for_every_leaf() {
556        let rows = certify(&MemoryBlob::new(), &Config::new(), &Config::new()).await;
557        assert_eq!(rows.len(), Leaf::ALL.len());
558        for c in &rows {
559            assert!(c.row.enabled, "{}: {}", c.leaf.id(), c.row.note);
560            assert!(c.row.metadata.get("test").is_some());
561        }
562    }
563
564    #[tokio::test]
565    async fn every_defect_turns_its_own_leaf_off_as_wrong() {
566        for leaf in Leaf::ALL {
567            for d in leaf.defects() {
568                let rows =
569                    certify(&MemoryBlob::with_defect(*d), &Config::new(), &Config::new()).await;
570                let mine = rows.iter().find(|c| c.leaf == leaf).unwrap();
571                assert!(!mine.row.enabled, "{d:?} must turn {} off", leaf.id());
572                assert_eq!(mine.row.reason, Some(Reason::Wrong), "{d:?}");
573            }
574        }
575    }
576
577    #[tokio::test]
578    async fn the_harness_can_tell_leaves_apart() {
579        // Control: ReadTruncates breaks blob.read and leaves blob.delete on.
580        let rows = certify(
581            &MemoryBlob::with_defect(Defect::ReadTruncates),
582            &Config::new(),
583            &Config::new(),
584        )
585        .await;
586        let by = |l: Leaf| rows.iter().find(|c| c.leaf == l).unwrap().row.enabled;
587        assert!(!by(Leaf::Read));
588        assert!(by(Leaf::Stat), "a read defect must not turn stat off");
589    }
590
591    #[tokio::test]
592    async fn rows_round_trip_through_the_matrix_and_the_gate_reads_them() {
593        let rows: Vec<Row> = certify(&MemoryBlob::new(), &Config::new(), &Config::new())
594            .await
595            .into_iter()
596            .map(|c| c.row)
597            .collect();
598        let m = persistant_core::Matrix::parse(&persistant_core::Matrix::from_rows(rows).to_tsv())
599            .unwrap();
600        let id = CapabilityId::new("blob.list").unwrap();
601        let v = env!("CARGO_PKG_VERSION");
602        assert!(m.enabled(&id, "memory", v, &Config::new()));
603        assert!(
604            !m.enabled(&id, "memory", "0.0.0-other", &Config::new()),
605            "unknown version is off"
606        );
607    }
608}
persistant_blob/conformance.rs

persistant_blob/
conformance.rs
