persistant_blob/
memory.rs1use crate::concurrency::yield_now;
3use crate::{BlobBackend, BlobError, Entry, Page};
4use std::collections::BTreeMap;
5use std::sync::Mutex;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11#[non_exhaustive]
12pub enum Defect {
13 ReadTruncates,
15 KeysFoldCase,
17 WriteIsLost,
19 DeleteIsIgnored,
21 StatLiesAboutLength,
23 ListIsDescending,
25 ListPrefixIsPattern,
27 ListIgnoresPaging,
29 CreateNotExclusive,
31 AppendOverwrites,
33 AppendTearsRecords,
35 ReplaceTearsReads,
38}
39
40#[derive(Debug, Default)]
42pub struct MemoryBlob {
43 map: Mutex<BTreeMap<String, Vec<u8>>>,
44 defect: Option<Defect>,
45}
46
47impl MemoryBlob {
48 pub fn new() -> Self {
50 MemoryBlob::default()
51 }
52
53 pub fn with_defect(defect: Defect) -> Self {
55 MemoryBlob {
56 map: Mutex::default(),
57 defect: Some(defect),
58 }
59 }
60
61 fn k(&self, key: &str) -> String {
62 if self.defect == Some(Defect::KeysFoldCase) {
63 key.to_lowercase()
64 } else {
65 key.to_string()
66 }
67 }
68
69 fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> {
70 self.map.lock().unwrap_or_else(|e| e.into_inner())
72 }
73}
74
75fn pattern_prefix_match(key: &str, pattern: &str) -> bool {
78 let cut = pattern.find(['%', '[']).unwrap_or(pattern.len());
79 key.starts_with(&pattern[..cut])
80}
81
82impl BlobBackend for MemoryBlob {
83 fn backend(&self) -> &str {
84 "memory"
85 }
86
87 fn version(&self) -> &str {
88 env!("CARGO_PKG_VERSION")
89 }
90
91 async fn read(&self, key: &str) -> Result<Option<Vec<u8>>, BlobError> {
92 let mut v = self.map().get(&self.k(key)).cloned();
93 if self.defect == Some(Defect::ReadTruncates)
94 && let Some(b) = v.as_mut()
95 {
96 b.pop();
97 }
98 Ok(v)
99 }
100
101 async fn write(&self, key: &str, bytes: Vec<u8>) -> Result<(), BlobError> {
102 if self.defect != Some(Defect::WriteIsLost) {
103 self.map().insert(self.k(key), bytes);
104 }
105 Ok(())
106 }
107
108 async fn delete(&self, key: &str) -> Result<(), BlobError> {
109 if self.defect != Some(Defect::DeleteIsIgnored) {
110 self.map().remove(key);
111 }
112 Ok(())
113 }
114
115 async fn stat(&self, key: &str) -> Result<Option<Entry>, BlobError> {
116 let extra = u64::from(self.defect == Some(Defect::StatLiesAboutLength));
117 Ok(self
118 .map()
119 .get(key)
120 .map(|v| Entry::new(key, v.len() as u64 + extra)))
121 }
122
123 async fn list(
124 &self,
125 prefix: &str,
126 after: Option<&str>,
127 limit: usize,
128 ) -> Result<Page, BlobError> {
129 let map = self.map();
130 let glob = self.defect == Some(Defect::ListPrefixIsPattern);
131 let ignore_paging = self.defect == Some(Defect::ListIgnoresPaging);
132 let mut keys: Vec<&String> = map
133 .keys()
134 .filter(|k| {
135 if glob {
136 pattern_prefix_match(k, prefix)
137 } else {
138 k.starts_with(prefix)
139 }
140 })
141 .filter(|k| ignore_paging || after.is_none_or(|a| k.as_str() > a))
142 .collect();
143 if self.defect == Some(Defect::ListIsDescending) {
144 keys.reverse();
145 }
146 let limit = if ignore_paging { usize::MAX } else { limit };
147 let more = keys.len() > limit;
148 let entries: Vec<Entry> = keys
149 .into_iter()
150 .take(limit)
151 .map(|k| Entry::new(k.clone(), map[k].len() as u64))
152 .collect();
153 let next = if more {
154 entries.last().map(|e| e.key.clone())
155 } else {
156 None
157 };
158 Ok(Page::new(entries, next))
159 }
160
161 async fn create_exclusive(&self, key: &str, bytes: Vec<u8>) -> Result<bool, BlobError> {
162 if self.defect == Some(Defect::CreateNotExclusive) {
163 let absent = !self.map().contains_key(key);
164 yield_now().await;
165 if absent {
166 self.map().insert(key.to_string(), bytes);
167 }
168 return Ok(absent);
169 }
170 let mut map = self.map();
171 if map.contains_key(key) {
172 return Ok(false);
173 }
174 map.insert(key.to_string(), bytes);
175 Ok(true)
176 }
177
178 async fn replace_atomic(&self, key: &str, bytes: Vec<u8>) -> Result<(), BlobError> {
179 if self.defect == Some(Defect::ReplaceTearsReads) {
180 let half = bytes.len() / 2;
181 let mut torn = bytes[..half].to_vec();
182 if let Some(old) = self.map().get(key) {
183 torn.extend_from_slice(old.get(half..).unwrap_or(&[]));
184 } else {
185 torn.extend_from_slice(&bytes[half..]);
186 }
187 self.map().insert(key.to_string(), torn);
188 yield_now().await;
189 }
190 self.map().insert(key.to_string(), bytes);
191 Ok(())
192 }
193
194 async fn append(&self, key: &str, bytes: Vec<u8>) -> Result<(), BlobError> {
195 match self.defect {
196 Some(Defect::AppendOverwrites) => {
197 self.map().insert(key.to_string(), bytes);
198 }
199 Some(Defect::AppendTearsRecords) => {
200 let half = bytes.len() / 2;
201 self.map()
202 .entry(key.to_string())
203 .or_default()
204 .extend_from_slice(&bytes[..half]);
205 yield_now().await;
206 self.map()
207 .entry(key.to_string())
208 .or_default()
209 .extend_from_slice(&bytes[half..]);
210 }
211 _ => {
212 self.map()
213 .entry(key.to_string())
214 .or_default()
215 .extend_from_slice(&bytes);
216 }
217 }
218 Ok(())
219 }
220}