mnesis_store_testing/
lifecycle.rs

1//! Lifecycle conformance (opt-in): close → reopen must preserve events,
2//! versions, and the `$all` position watermark. In-memory adapters have
3//! nothing to reopen and skip this module.
4
5use core::future::Future;
6use core::time::Duration;
7
8use futures::StreamExt;
9use futures::pin_mut;
10use mnesis::Version;
11use mnesis_store::store::RawEventStore;
12use mnesis_store::wake::WakeSource;
13use mnesis_store::{PendingBatch, Step, StreamKey, Subscription};
14use tokio::time::timeout;
15
16use crate::row::{
17    ConformanceRow, SubId, append_event, append_rows, drain_all, drain_stream, envelope_for,
18};
19
20/// Everything written before close reads back identically after reopen.
21pub async fn check_reopen_preserves_events<S, C, O, OFut, R, RFut>(open: &O, reopen: &R)
22where
23    S: RawEventStore + WakeSource,
24    C: Send,
25    O: Fn() -> OFut + Send + Sync,
26    OFut: Future<Output = (S, C)> + Send,
27    R: Fn(S, C) -> RFut + Send + Sync,
28    RFut: Future<Output = (S, C)> + Send,
29{
30    let (opened, ctx) = open().await;
31    let id = StreamKey::from_slice(b"persist");
32    let rows = vec![
33        ConformanceRow::new(1, "Created", vec![1]).with_metadata(vec![7]),
34        ConformanceRow::new(2, "Updated", vec![2]).with_schema_version(3),
35    ];
36    append_rows(&opened, &id, &rows).await;
37    let before = drain_stream(&opened, &id, Version::INITIAL).await;
38
39    let (reopened, _ctx) = reopen(opened, ctx).await;
40    let after = drain_stream(&reopened, &id, Version::INITIAL).await;
41    assert_eq!(
42        after, before,
43        "reopen must preserve every event byte-for-byte"
44    );
45}
46
47/// The `$all` position watermark survives reopen: a post-reopen append lands
48/// strictly after every pre-close position (a reset counter would violate
49/// resume and corrupt projections).
50pub async fn check_reopen_preserves_position_watermark<S, C, O, OFut, R, RFut>(open: &O, reopen: &R)
51where
52    S: RawEventStore + WakeSource,
53    C: Send,
54    O: Fn() -> OFut + Send + Sync,
55    OFut: Future<Output = (S, C)> + Send,
56    R: Fn(S, C) -> RFut + Send + Sync,
57    RFut: Future<Output = (S, C)> + Send,
58{
59    let (opened, ctx) = open().await;
60    let a = StreamKey::from_slice(b"a");
61    let b = StreamKey::from_slice(b"b");
62    append_event(&opened, &a, 1, b"a1").await;
63    append_event(&opened, &b, 1, b"b1").await;
64    let before = drain_all(&opened, None).await;
65    let last = before.last().expect("non-empty").0;
66
67    let (reopened, _ctx) = reopen(opened, ctx).await;
68    append_event(&reopened, &a, 2, b"a2").await;
69
70    let resumed = drain_all(&reopened, Some(last)).await;
71    assert_eq!(
72        resumed.len(),
73        1,
74        "resume from the pre-close watermark must yield only the new event"
75    );
76    assert_eq!(resumed[0].1, b"a2".to_vec());
77    assert!(
78        resumed[0].0 > last,
79        "post-reopen position {:?} must be strictly after the pre-close last {:?} — the watermark must survive reopen",
80        resumed[0].0,
81        last,
82    );
83}
84
85/// Optimistic-concurrency state survives reopen: a stale expectation still
86/// conflicts against the persisted head.
87pub async fn check_reopen_conflict_state_intact<S, C, O, OFut, R, RFut>(open: &O, reopen: &R)
88where
89    S: RawEventStore + WakeSource,
90    C: Send,
91    O: Fn() -> OFut + Send + Sync,
92    OFut: Future<Output = (S, C)> + Send,
93    R: Fn(S, C) -> RFut + Send + Sync,
94    RFut: Future<Output = (S, C)> + Send,
95{
96    let (opened, ctx) = open().await;
97    let id = StreamKey::from_slice(b"head");
98    append_event(&opened, &id, 1, b"p1").await;
99    append_event(&opened, &id, 2, b"p2").await;
100
101    let (reopened, _ctx) = reopen(opened, ctx).await;
102    let stale = envelope_for(&ConformanceRow::new(1, "E", vec![9]));
103    reopened
104        .append(&id, None, PendingBatch::of(&stale))
105        .await
106        .expect_err("the persisted head must still conflict after reopen");
107    // The corrected append succeeds.
108    append_event(&reopened, &id, 3, b"p3").await;
109    let got = drain_stream(&reopened, &id, Version::INITIAL).await;
110    let versions: Vec<u64> = got.iter().map(|r| r.version).collect();
111    assert_eq!(versions, vec![1, 2, 3]);
112}
113
114/// A subscription opened after reopen catches up over the pre-close backlog.
115pub async fn check_reopen_subscription_catches_up<S, C, O, OFut, R, RFut>(open: &O, reopen: &R)
116where
117    S: RawEventStore + WakeSource,
118    S::Stream: Unpin,
119    C: Send,
120    O: Fn() -> OFut + Send + Sync,
121    OFut: Future<Output = (S, C)> + Send,
122    R: Fn(S, C) -> RFut + Send + Sync,
123    RFut: Future<Output = (S, C)> + Send,
124{
125    let (opened, ctx) = open().await;
126    let id = SubId::new("reopen-sub");
127    for v in 1..=3u64 {
128        append_event(&opened, &id.key(), v, b"p").await;
129    }
130
131    let (reopened, _ctx) = reopen(opened, ctx).await;
132    let store = reopened.into_store();
133    let sub = Subscription::new(&store);
134    let stream = sub
135        .subscribe(&id, None)
136        .unwrap_or_else(|e| panic!("register failed: {e:?}"));
137    pin_mut!(stream);
138
139    let mut versions = Vec::new();
140    while let Step::Event(env) = timeout(Duration::from_secs(10), stream.next())
141        .await
142        .expect("catch-up after reopen must not hang")
143        .expect("subscription must not end")
144        .unwrap_or_else(|e| panic!("item errored: {e:?}"))
145    {
146        versions.push(env.version().as_u64());
147    }
148    assert_eq!(versions, vec![1, 2, 3], "reopen backlog must replay fully");
149}
mnesis_store_testing/lifecycle.rs

mnesis_store_testing/
lifecycle.rs
