mnesis_store_testing/
atomic.rs

1//! `AtomicAppend` capability conformance: several per-stream runs commit in
2//! ONE transaction — all land or none do.
3
4use core::future::Future;
5
6use mnesis::Version;
7use mnesis_store::StreamKey;
8use mnesis_store::import::{AtomicAppend, AtomicAppendError, PlannedAppend};
9use mnesis_store::wake::WakeSource;
10// NOTE: RawEventStore is NOT imported — `AtomicAppend: RawEventStore` is a
11// supertrait, and nothing here names the trait directly (unused imports deny).
12
13use crate::row::{
14    ConformanceRow, append_rows, assert_strictly_increasing, drain_all, drain_stream, envelope_for,
15};
16
17/// Three runs across three streams (two fresh, one existing) commit together.
18pub async fn check_atomic_multi_stream_commits_all<S, C, F, Fut>(factory: &F)
19where
20    S: AtomicAppend + WakeSource,
21    C: Send,
22    F: Fn() -> Fut + Send + Sync,
23    Fut: Future<Output = (S, C)> + Send,
24{
25    let (store, _guard) = factory().await;
26    let existing = StreamKey::from_slice(b"existing");
27    append_rows(&store, &existing, &[ConformanceRow::new(1, "E", vec![0])]).await;
28
29    let writes = vec![
30        PlannedAppend {
31            target: StreamKey::from_slice(b"fresh-a"),
32            expected_version: None,
33            head: envelope_for(&ConformanceRow::new(1, "E", vec![1])),
34            tail: Vec::new(),
35        },
36        PlannedAppend {
37            target: StreamKey::from_slice(b"fresh-b"),
38            expected_version: None,
39            head: envelope_for(&ConformanceRow::new(1, "E", vec![2])),
40            tail: vec![envelope_for(&ConformanceRow::new(2, "E", vec![3]))],
41        },
42        PlannedAppend {
43            target: existing.clone(),
44            expected_version: Version::new(1),
45            head: envelope_for(&ConformanceRow::new(2, "E", vec![4])),
46            tail: Vec::new(),
47        },
48    ];
49    let committed = store
50        .atomic_append_many(&writes)
51        .await
52        .unwrap_or_else(|e| panic!("atomic append must succeed: {e:?}"));
53
54    assert_eq!(
55        drain_stream(&store, &StreamKey::from_slice(b"fresh-a"), Version::INITIAL)
56            .await
57            .len(),
58        1
59    );
60    assert_eq!(
61        drain_stream(&store, &StreamKey::from_slice(b"fresh-b"), Version::INITIAL)
62            .await
63            .len(),
64        2
65    );
66    assert_eq!(
67        drain_stream(&store, &existing, Version::INITIAL)
68            .await
69            .len(),
70        2
71    );
72    let all = drain_all(&store, None).await;
73    assert_eq!(all.len(), 5, "$all must hold every committed event");
74    assert_strictly_increasing(&all);
75
76    // The returned position is the read-your-writes token for the WHOLE
77    // transaction (#330): a consumer that has reached it has been delivered
78    // every event the batch committed, across every stream it touched.
79    let highest = all
80        .iter()
81        .map(|(pos, _)| *pos)
82        .max()
83        .expect("the batch committed events");
84    assert_eq!(
85        committed,
86        Some(highest),
87        "atomic_append_many must return the highest position it committed"
88    );
89}
90
91/// A conflict in ONE run aborts the WHOLE batch: no stream changes, the error
92/// names the offending write index and the actual head.
93pub async fn check_atomic_conflict_aborts_all<S, C, F, Fut>(factory: &F)
94where
95    S: AtomicAppend + WakeSource,
96    C: Send,
97    F: Fn() -> Fut + Send + Sync,
98    Fut: Future<Output = (S, C)> + Send,
99{
100    let (store, _guard) = factory().await;
101    let existing = StreamKey::from_slice(b"existing");
102    append_rows(&store, &existing, &[ConformanceRow::new(1, "E", vec![0])]).await;
103    let all_before = drain_all(&store, None).await;
104
105    let writes = vec![
106        PlannedAppend {
107            target: StreamKey::from_slice(b"fresh-a"),
108            expected_version: None,
109            head: envelope_for(&ConformanceRow::new(1, "E", vec![1])),
110            tail: Vec::new(),
111        },
112        PlannedAppend {
113            // WRONG: head is 1, we claim fresh.
114            target: existing.clone(),
115            expected_version: None,
116            head: envelope_for(&ConformanceRow::new(1, "E", vec![9])),
117            tail: Vec::new(),
118        },
119    ];
120    let err = store
121        .atomic_append_many(&writes)
122        .await
123        .expect_err("a conflicting run must abort the batch");
124    match err {
125        AtomicAppendError::Conflict { index, actual } => {
126            assert_eq!(index, 1, "the error must name the offending write");
127            assert_eq!(
128                actual,
129                Version::new(1),
130                "the error must carry the actual head"
131            );
132        }
133        other => panic!("expected Conflict, got {other:?}"),
134    }
135
136    let fresh = drain_stream(&store, &StreamKey::from_slice(b"fresh-a"), Version::INITIAL).await;
137    assert!(
138        fresh.is_empty(),
139        "NOTHING may land on any stream of an aborted batch"
140    );
141    let all_after = drain_all(&store, None).await;
142    assert_eq!(
143        all_after.len(),
144        all_before.len(),
145        "$all must be untouched by an aborted batch"
146    );
147}
148
149/// An empty batch is a no-op `Ok` — and commits no position.
150pub async fn check_atomic_empty_batch_is_noop<S, C, F, Fut>(factory: &F)
151where
152    S: AtomicAppend + WakeSource,
153    C: Send,
154    F: Fn() -> Fut + Send + Sync,
155    Fut: Future<Output = (S, C)> + Send,
156{
157    let (store, _guard) = factory().await;
158    let committed = store
159        .atomic_append_many(&[])
160        .await
161        .unwrap_or_else(|e| panic!("empty atomic batch must be Ok: {e:?}"));
162    assert_eq!(
163        committed, None,
164        "an empty batch commits nothing, so there is no position to return"
165    );
166    assert!(
167        drain_all(&store, None).await.is_empty(),
168        "empty batch must write nothing"
169    );
170}
mnesis_store_testing/atomic.rs

mnesis_store_testing/
atomic.rs
