matrix_sdk_base/event_cache/store/
controlled_memory_store.rs

1// Copyright 2026 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! A [`MemoryStore`] whose writes can be blocked or made to fail, for tests.
16
17use std::{
18    sync::{
19        Arc,
20        atomic::{AtomicBool, Ordering},
21    },
22    time::Duration,
23};
24
25use async_trait::async_trait;
26use matrix_sdk_common::{
27    cross_process_lock::CrossProcessLockGeneration,
28    linked_chunk::{
29        ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position,
30        RawChunk, Update,
31    },
32    sleep::sleep,
33};
34use ruma::{EventId, OwnedEventId, RoomId, events::relation::RelationType};
35use tokio::sync::Notify;
36
37use super::{
38    super::{Event, Gap, thread::ThreadInfo},
39    EventCacheStore, EventCacheStoreError, MemoryStore, Result,
40};
41
42/// A [`MemoryStore`] where we can control which operations succeed or fail.
43///
44/// This is meant for tests that need to control when, or whether, the event
45/// cache persists its data: for instance to check what the rest of the SDK
46/// does while the store is slow, or when it returns an error.
47///
48/// Only [`EventCacheStore::handle_linked_chunk_updates`] and
49/// [`EventCacheStore::load_last_chunk`] are affected. All the other methods are
50/// forwarded to the inner [`MemoryStore`] as is.
51#[derive(Debug, Clone, Default)]
52pub struct ControlledMemoryStore {
53    inner: MemoryStore,
54    writes_blocked: Arc<AtomicBool>,
55    writes_failing: Arc<AtomicBool>,
56    fail_last_chunk_load: Arc<AtomicBool>,
57    write_started: Arc<Notify>,
58    write_finished: Arc<Notify>,
59}
60
61impl ControlledMemoryStore {
62    /// Create a new store, which neither blocks nor fails writes.
63    pub fn new() -> Self {
64        Self::default()
65    }
66
67    /// Block linked chunk writes until [`Self::unblock_writes`] is called.
68    ///
69    /// A blocked write still counts as started for
70    /// [`Self::wait_for_write_to_start`].
71    pub fn block_writes(&self) {
72        self.writes_blocked.store(true, Ordering::SeqCst);
73    }
74
75    /// Let the blocked linked chunk writes, and the next ones, go through.
76    pub fn unblock_writes(&self) {
77        self.writes_blocked.store(false, Ordering::SeqCst);
78    }
79
80    /// Make linked chunk writes fail with an [`EventCacheStoreError::Backend`]
81    /// error until [`Self::stop_failing_writes`] is called.
82    ///
83    /// A failing write doesn't touch the inner [`MemoryStore`].
84    pub fn fail_writes(&self) {
85        self.writes_failing.store(true, Ordering::SeqCst);
86    }
87
88    /// Make linked chunk writes succeed again.
89    pub fn stop_failing_writes(&self) {
90        self.writes_failing.store(false, Ordering::SeqCst);
91    }
92
93    /// Make calls to [`EventCacheStore::load_last_chunk`] fail.
94    pub fn fail_next_last_chunk_load(&self) {
95        self.fail_last_chunk_load.store(true, Ordering::SeqCst);
96    }
97
98    /// Stop making calls to [`EventCacheStore::load_last_chunk`] fail.
99    pub fn stop_failing_next_last_chunk_load(&self) {
100        self.fail_last_chunk_load.store(false, Ordering::SeqCst);
101    }
102
103    /// Wait until a linked chunk write starts.
104    ///
105    /// Returns at once if a write has started since the last call.
106    pub async fn wait_for_write_to_start(&self) {
107        self.write_started.notified().await;
108    }
109
110    /// Wait until a linked chunk write finishes, with or without an error.
111    ///
112    /// Returns at once if a write has finished since the last call.
113    pub async fn wait_for_write_to_finish(&self) {
114        self.write_finished.notified().await;
115    }
116}
117
118#[cfg_attr(target_family = "wasm", async_trait(?Send))]
119#[cfg_attr(not(target_family = "wasm"), async_trait)]
120impl EventCacheStore for ControlledMemoryStore {
121    type Error = EventCacheStoreError;
122
123    async fn handle_linked_chunk_updates(
124        &self,
125        linked_chunk_id: LinkedChunkId<'_>,
126        updates: Vec<Update<Event, Gap>>,
127    ) -> Result<(), Self::Error> {
128        self.write_started.notify_one();
129
130        // Polling is good enough for tests, and simpler than a proper wake up.
131        while self.writes_blocked.load(Ordering::SeqCst) {
132            sleep(Duration::from_millis(10)).await;
133        }
134
135        let result = if self.writes_failing.load(Ordering::SeqCst) {
136            Err(EventCacheStoreError::backend(std::io::Error::other(
137                "the event cache store is failing on purpose",
138            )))
139        } else {
140            self.inner.handle_linked_chunk_updates(linked_chunk_id, updates).await
141        };
142
143        self.write_finished.notify_one();
144
145        result
146    }
147
148    async fn close(&self) -> Result<(), Self::Error> {
149        self.inner.close().await
150    }
151
152    async fn reopen(&self) -> Result<(), Self::Error> {
153        self.inner.reopen().await
154    }
155
156    async fn try_take_leased_lock(
157        &self,
158        lease_duration_ms: u32,
159        key: &str,
160        holder: &str,
161    ) -> Result<Option<CrossProcessLockGeneration>, Self::Error> {
162        self.inner.try_take_leased_lock(lease_duration_ms, key, holder).await
163    }
164
165    async fn load_all_chunks(
166        &self,
167        linked_chunk_id: LinkedChunkId<'_>,
168    ) -> Result<Vec<RawChunk<Event, Gap>>, Self::Error> {
169        self.inner.load_all_chunks(linked_chunk_id).await
170    }
171
172    async fn load_all_chunks_metadata(
173        &self,
174        linked_chunk_id: LinkedChunkId<'_>,
175    ) -> Result<Vec<ChunkMetadata>, Self::Error> {
176        self.inner.load_all_chunks_metadata(linked_chunk_id).await
177    }
178
179    async fn load_last_chunk(
180        &self,
181        linked_chunk_id: LinkedChunkId<'_>,
182    ) -> Result<(Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator), Self::Error> {
183        if self.fail_last_chunk_load.load(Ordering::SeqCst) {
184            return Err(EventCacheStoreError::backend(std::io::Error::other(
185                "`load_last_chunk` failure requested by the test",
186            )));
187        }
188
189        self.inner.load_last_chunk(linked_chunk_id).await
190    }
191
192    async fn load_previous_chunk(
193        &self,
194        linked_chunk_id: LinkedChunkId<'_>,
195        before_chunk_identifier: ChunkIdentifier,
196    ) -> Result<Option<RawChunk<Event, Gap>>, Self::Error> {
197        self.inner.load_previous_chunk(linked_chunk_id, before_chunk_identifier).await
198    }
199
200    async fn load_thread_info(
201        &self,
202        room_id: &RoomId,
203        thread_id: &EventId,
204        insert_default_if_missing: bool,
205    ) -> Result<Option<ThreadInfo>, Self::Error> {
206        self.inner.load_thread_info(room_id, thread_id, insert_default_if_missing).await
207    }
208
209    async fn update_thread_info(
210        &self,
211        room_id: &RoomId,
212        thread_id: &EventId,
213        thread_info: &ThreadInfo,
214    ) -> Result<(), Self::Error> {
215        self.inner.update_thread_info(room_id, thread_id, thread_info).await
216    }
217
218    async fn clear_all_events(&self, room_id: Option<&RoomId>) -> Result<(), Self::Error> {
219        self.inner.clear_all_events(room_id).await
220    }
221
222    async fn filter_duplicated_events(
223        &self,
224        linked_chunk_id: LinkedChunkId<'_>,
225        events: Vec<OwnedEventId>,
226    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error> {
227        self.inner.filter_duplicated_events(linked_chunk_id, events).await
228    }
229
230    async fn find_event(
231        &self,
232        room_id: &RoomId,
233        event_id: &EventId,
234    ) -> Result<Option<Event>, Self::Error> {
235        self.inner.find_event(room_id, event_id).await
236    }
237
238    async fn find_event_relations(
239        &self,
240        room_id: &RoomId,
241        event_id: &EventId,
242        filters: Option<&[RelationType]>,
243    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
244        self.inner.find_event_relations(room_id, event_id, filters).await
245    }
246
247    async fn get_room_events(
248        &self,
249        room_id: &RoomId,
250        event_type: Option<&str>,
251        session_id: Option<&str>,
252    ) -> Result<Vec<Event>, Self::Error> {
253        self.inner.get_room_events(room_id, event_type, session_id).await
254    }
255
256    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {
257        self.inner.save_event(room_id, event).await
258    }
259
260    async fn optimize(&self) -> Result<(), Self::Error> {
261        self.inner.optimize().await
262    }
263
264    async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
265        self.inner.get_size().await
266    }
267
268    async fn find_events_before_timestamp(
269        &self,
270        room_id: &RoomId,
271        cutoff_ms: u64,
272    ) -> Result<Vec<Event>, Self::Error> {
273        self.inner.find_events_before_timestamp(room_id, cutoff_ms).await
274    }
275}
matrix_sdk_base/event_cache/store/controlled_memory_store.rs

matrix_sdk_base/event_cache/store/
controlled_memory_store.rs
