1use super::event::SessionEvent;
2use super::manager::Session;
3use std::{
4 collections::VecDeque,
5 fs,
6 io::{BufRead, BufReader, Read, Seek, SeekFrom},
7 path::PathBuf,
8 time::SystemTime,
9};
10
11pub fn validate_session_id(id: String) -> anyhow::Result<String> {
12 if id.is_empty() {
13 anyhow::bail!("session id must not be empty");
14 }
15 if id == "." || id == ".." || id.contains("..") {
16 anyhow::bail!("session id must not contain '..'");
17 }
18 if id.contains('/') || id.contains('\\') {
19 anyhow::bail!("session id must not contain path separators");
20 }
21 if PathBuf::from(&id).is_absolute() {
22 anyhow::bail!("session id must not be an absolute path");
23 }
24 if !id
25 .chars()
26 .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-')
27 {
28 anyhow::bail!("session id must match [A-Za-z0-9_-]+");
29 }
30 Ok(id)
31}
32
33fn open_session_file(session: &Session) -> anyhow::Result<fs::File> {
34 let root = session
35 .path
36 .parent()
37 .ok_or_else(|| anyhow::anyhow!("session file has no parent"))?;
38 super::store::open_existing_primary(root, &session.id)?
39 .ok_or_else(|| anyhow::anyhow!("session JSONL is missing"))
40}
41#[cfg(test)]
42pub(crate) fn latest_valid_event_timestamp_streaming(session: &Session) -> Option<SystemTime> {
43 let root = session.path.parent()?;
44 let file = super::store::open_existing_primary(root, &session.id).ok()??;
45 let mut latest = None;
46 for line in BufReader::new(file).lines() {
47 let Ok(line) = line else {
48 continue;
49 };
50 if line.trim().is_empty() {
51 continue;
52 }
53 let Ok(event) = serde_json::from_str::<SessionEvent>(&line) else {
54 continue;
55 };
56 let timestamp = SystemTime::from(event.timestamp);
57 latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
58 }
59 latest
60}
61
62#[derive(Debug)]
63pub(crate) enum BoundedReadError {
64 BudgetExceeded(String),
65}
66
67impl std::fmt::Display for BoundedReadError {
68 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69 match self {
70 Self::BudgetExceeded(message) => formatter.write_str(message),
71 }
72 }
73}
74
75impl std::error::Error for BoundedReadError {}
76
77fn budget_error(message: impl Into<String>) -> anyhow::Error {
78 anyhow::Error::new(BoundedReadError::BudgetExceeded(message.into()))
79}
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct SessionReadDiagnostic {
82 pub line: usize,
83 pub message: String,
84}
85
86#[derive(Debug, Clone, PartialEq)]
87pub struct TolerantSessionEvents {
88 pub events: Vec<SessionEvent>,
89 pub diagnostics: Vec<SessionReadDiagnostic>,
90 pub(crate) cutoff_bytes: u64,
91}
92
93const MAX_TOLERANT_READ_DIAGNOSTICS: usize = 64;
94
95fn push_tolerant_read_diagnostic(
96 diagnostics: &mut Vec<SessionReadDiagnostic>,
97 omitted_count: &mut usize,
98 diagnostic: SessionReadDiagnostic,
99) {
100 if diagnostics.len() < MAX_TOLERANT_READ_DIAGNOSTICS {
101 diagnostics.push(diagnostic);
102 } else {
103 *omitted_count = omitted_count.saturating_add(1);
104 }
105}
106
107fn finalize_tolerant_read_diagnostics(
108 mut diagnostics: Vec<SessionReadDiagnostic>,
109 omitted_count: usize,
110) -> Vec<SessionReadDiagnostic> {
111 if omitted_count == 0 {
112 return diagnostics;
113 }
114 if diagnostics.len() == MAX_TOLERANT_READ_DIAGNOSTICS {
115 diagnostics.pop();
116 }
117 diagnostics.push(SessionReadDiagnostic {
118 line: 0,
119 message: format!(
120 "omitted {omitted_count} additional session JSONL diagnostics after cap of {MAX_TOLERANT_READ_DIAGNOSTICS}"
121 ),
122 });
123 diagnostics
124}
125
126pub(crate) const MAX_METADATA_VISIT_LINES: usize = 100_000;
127pub(crate) const MAX_METADATA_VISIT_BYTES: usize = 64 * 1024 * 1024;
128
129impl Session {
130 pub fn read_events(&self) -> anyhow::Result<Vec<SessionEvent>> {
131 validate_session_id(self.id.clone())?;
132 if !self.path.exists() {
133 return Ok(Vec::new());
134 }
135 self.read_event_lines_streaming()?
136 .map(|event_line| event_line.map(|(event, _)| event))
137 .collect()
138 }
139
140 pub fn read_recent_events(
141 &self,
142 max_events: usize,
143 max_bytes: usize,
144 ) -> anyhow::Result<Vec<SessionEvent>> {
145 validate_session_id(self.id.clone())?;
146 if !self.path.exists() || max_events == 0 || max_bytes == 0 {
147 return Ok(Vec::new());
148 }
149 let mut retained = VecDeque::new();
150 let mut retained_bytes = 0usize;
151 for event_line in self.read_event_lines_streaming()? {
152 let (event, line_bytes) = event_line?;
153 retained_bytes = retained_bytes.saturating_add(line_bytes);
155 retained.push_back((event, line_bytes));
156 while retained.len() > max_events || retained_bytes > max_bytes {
157 if let Some((_, bytes)) = retained.pop_front() {
158 retained_bytes = retained_bytes.saturating_sub(bytes);
159 } else {
160 break;
161 }
162 }
163 }
164 Ok(retained.into_iter().map(|(event, _)| event).collect())
165 }
166
167 pub fn read_events_tolerant(&self) -> anyhow::Result<TolerantSessionEvents> {
168 self.read_events_tolerant_bounded(usize::MAX, usize::MAX)
169 }
170
171 pub(crate) fn read_events_tolerant_bounded(
172 &self,
173 max_lines: usize,
174 max_bytes: usize,
175 ) -> anyhow::Result<TolerantSessionEvents> {
176 let mut events = Vec::new();
177 let (diagnostics, cutoff_bytes) =
178 self.visit_events_tolerant_bounded(max_lines, max_bytes, |event| events.push(event))?;
179 Ok(TolerantSessionEvents {
180 events,
181 diagnostics,
182 cutoff_bytes: cutoff_bytes as u64,
183 })
184 }
185
186 pub(crate) fn visit_events_tolerant_bounded(
187 &self,
188 max_lines: usize,
189 max_bytes: usize,
190 mut visit: impl FnMut(SessionEvent),
191 ) -> anyhow::Result<(Vec<SessionReadDiagnostic>, usize)> {
192 validate_session_id(self.id.clone())?;
193 if !self.path.exists() {
194 return Ok((Vec::new(), 0));
195 }
196 let file = open_session_file(self)?;
197 let mut reader = BufReader::new(file);
198 let mut diagnostics = Vec::new();
199 let mut omitted_diagnostics = 0usize;
200 let mut total_bytes = 0usize;
201 let mut line = Vec::new();
202 for line_number in 1..=max_lines {
203 line.clear();
204 let remaining = max_bytes.saturating_sub(total_bytes);
205 if remaining == 0 {
206 if !reader.fill_buf()?.is_empty() {
207 return Err(budget_error(format!(
208 "session JSONL tolerant read limit exceeded: {max_bytes} bytes"
209 )));
210 }
211 break;
212 }
213 let read = (&mut reader)
214 .take(
215 u64::try_from(remaining)
216 .unwrap_or(u64::MAX)
217 .saturating_add(1),
218 )
219 .read_until(b'\n', &mut line)?;
220 if read == 0 {
221 break;
222 }
223 if read > remaining {
224 return Err(budget_error(format!(
225 "session JSONL tolerant read limit exceeded at line {line_number}: {max_bytes} bytes"
226 )));
227 }
228 total_bytes = total_bytes.saturating_add(read);
229 if line.last() == Some(&b'\n') {
230 line.pop();
231 if line.last() == Some(&b'\r') {
232 line.pop();
233 }
234 }
235 let text = String::from_utf8_lossy(&line);
236 match serde_json::from_str::<SessionEvent>(&text) {
237 Ok(event) => visit(event),
238 Err(_) => push_tolerant_read_diagnostic(
239 &mut diagnostics,
240 &mut omitted_diagnostics,
241 SessionReadDiagnostic {
242 line: line_number,
243 message: format!(
244 "operation=replay category=session_jsonl failed to parse session JSONL at line {line_number}"
245 ),
246 },
247 ),
248 }
249 }
250 if !reader.fill_buf()?.is_empty() {
251 return Err(budget_error(format!(
252 "session JSONL tolerant read limit exceeded: more than {max_lines} lines or {max_bytes} bytes"
253 )));
254 }
255 Ok((
256 finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
257 total_bytes,
258 ))
259 }
260
261 pub fn read_recent_events_tolerant(
262 &self,
263 max_events: usize,
264 max_bytes: usize,
265 ) -> anyhow::Result<TolerantSessionEvents> {
266 validate_session_id(self.id.clone())?;
267 if !self.path.exists() || max_events == 0 || max_bytes == 0 {
268 return Ok(TolerantSessionEvents {
269 events: Vec::new(),
270 diagnostics: Vec::new(),
271 cutoff_bytes: 0,
272 });
273 }
274 let file = open_session_file(self)?;
275 let lines = BufReader::new(file)
276 .lines()
277 .enumerate()
278 .map(|(index, line)| (index + 1, line));
279 self.collect_recent_events_tolerant_lines(lines, max_events, max_bytes)
280 }
281
282 pub(crate) fn read_recent_events_tolerant_tail(
283 &self,
284 max_events: usize,
285 max_retained_bytes: usize,
286 max_read_bytes: usize,
287 ) -> anyhow::Result<TolerantSessionEvents> {
288 validate_session_id(self.id.clone())?;
289 if max_events == 0 || max_retained_bytes == 0 || max_read_bytes == 0 {
290 anyhow::bail!(
291 "session JSONL tolerant tail read limits must be non-zero: max_events={max_events}, max_retained_bytes={max_retained_bytes}, max_read_bytes={max_read_bytes}"
292 );
293 }
294 if !self.path.exists() {
295 return Ok(TolerantSessionEvents {
296 events: Vec::new(),
297 diagnostics: Vec::new(),
298 cutoff_bytes: 0,
299 });
300 }
301 let metadata = open_session_file(self)?.metadata()?;
302 let file_len = metadata.len();
303 let max_read_bytes_u64 = u64::try_from(max_read_bytes).unwrap_or(u64::MAX);
304 if file_len <= max_read_bytes_u64 {
305 return self.read_recent_events_tolerant(max_events, max_retained_bytes);
306 }
307 let start = file_len - max_read_bytes_u64;
308 let mut file = open_session_file(self)?;
309 file.seek(SeekFrom::Start(start))?;
310 let mut tail = Vec::with_capacity(max_read_bytes);
311 file.take(max_read_bytes_u64).read_to_end(&mut tail)?;
312
313 let first_complete_line = tail
314 .iter()
315 .position(|byte| *byte == b'\n')
316 .map_or(tail.len(), |index| index + 1);
317 let tail = String::from_utf8_lossy(&tail[first_complete_line..]);
318 let lines = tail
319 .lines()
320 .enumerate()
321 .map(|(index, line)| (index + 1, Ok(line.to_string())));
322 let mut result =
323 self.collect_recent_events_tolerant_lines(lines, max_events, max_retained_bytes)?;
324 result.diagnostics.insert(
325 0,
326 SessionReadDiagnostic {
327 line: 0,
328 message: format!(
329 "operation=recent_context category=session_jsonl bounded tail window; omitted older lines; read final {max_read_bytes} of {file_len} bytes"
330 ),
331 },
332 );
333 Ok(result)
334 }
335
336 fn collect_recent_events_tolerant_lines(
337 &self,
338 lines: impl IntoIterator<Item = (usize, Result<String, std::io::Error>)>,
339 max_events: usize,
340 max_bytes: usize,
341 ) -> anyhow::Result<TolerantSessionEvents> {
342 let mut retained = VecDeque::new();
343 let mut retained_bytes = 0usize;
344 let mut malformed_bytes = 0usize;
345 let mut diagnostics = Vec::new();
346 let mut omitted_diagnostics = 0usize;
347 for (line_number, line) in lines {
348 match line {
349 Ok(line) => {
350 let line_bytes = line.len() + 1;
351 match serde_json::from_str::<SessionEvent>(&line) {
352 Ok(event) => {
353 retained_bytes = retained_bytes.saturating_add(line_bytes);
354 retained.push_back((event, line_bytes));
355 while retained.len() > max_events || retained_bytes > max_bytes {
356 if let Some((_, bytes)) = retained.pop_front() {
357 retained_bytes = retained_bytes.saturating_sub(bytes);
358 } else {
359 break;
360 }
361 }
362 }
363 Err(_) => {
364 malformed_bytes = malformed_bytes.saturating_add(line_bytes);
365 if malformed_bytes > max_bytes {
366 anyhow::bail!(
367 "operation=recent_context category=session_jsonl tolerant recent read limit exceeded: malformed bytes > {max_bytes}"
368 );
369 }
370 push_tolerant_read_diagnostic(
371 &mut diagnostics,
372 &mut omitted_diagnostics,
373 SessionReadDiagnostic {
374 line: line_number,
375 message: format!(
376 "operation=recent_context category=session_jsonl failed to parse session JSONL at line {line_number}"
377 ),
378 },
379 );
380 }
381 }
382 }
383 Err(_) => push_tolerant_read_diagnostic(
384 &mut diagnostics,
385 &mut omitted_diagnostics,
386 SessionReadDiagnostic {
387 line: line_number,
388 message: format!(
389 "operation=recent_context category=session_jsonl read failure at line {line_number}"
390 ),
391 },
392 ),
393 }
394 }
395 Ok(TolerantSessionEvents {
396 events: retained.into_iter().map(|(event, _)| event).collect(),
397 diagnostics: finalize_tolerant_read_diagnostics(diagnostics, omitted_diagnostics),
398 cutoff_bytes: 0,
399 })
400 }
401
402 pub fn latest_event_timestamp_bounded(&self) -> anyhow::Result<Option<SystemTime>> {
403 validate_session_id(self.id.clone())?;
404 if !self.path.exists() {
405 return Ok(None);
406 }
407 let mut latest = None;
408 for event_line in self.read_event_lines_streaming()? {
409 let timestamp = SystemTime::from(event_line?.0.timestamp);
410 latest = Some(latest.map_or(timestamp, |current: SystemTime| current.max(timestamp)));
411 }
412 Ok(latest)
413 }
414
415 fn read_event_lines_streaming(
416 &self,
417 ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<(SessionEvent, usize)>> + '_> {
418 let file = open_session_file(self)?;
419 Ok(BufReader::new(file)
420 .lines()
421 .enumerate()
422 .map(|(index, line)| {
423 let line = line.map_err(|error| {
424 anyhow::anyhow!(
425 "failed to read session JSONL at {} line {}: {error}",
426 self.path.display(),
427 index + 1
428 )
429 })?;
430 let line_bytes = line.len() + 1;
432 let event = serde_json::from_str::<SessionEvent>(&line).map_err(|error| {
433 anyhow::anyhow!(
434 "failed to parse session JSONL at {} line {}: {error}",
435 self.path.display(),
436 index + 1
437 )
438 })?;
439 Ok((event, line_bytes))
440 }))
441 }
442}
443
444#[cfg(test)]
445mod tests {
446 use super::super::manager::SessionManager;
447 use super::*;
448 use proptest::prelude::*;
449 use serde_json::json;
450 use tempfile::TempDir;
451
452 fn valid_session_id_strategy() -> impl Strategy<Value = String> {
453 proptest::string::string_regex("[A-Za-z0-9_-]{1,64}").unwrap()
454 }
455
456 fn invalid_session_id_strategy() -> impl Strategy<Value = String> {
457 prop_oneof![
458 Just(String::new()),
459 Just(".".to_string()),
460 any::<String>().prop_map(|value| format!("{value}..")),
461 any::<String>().prop_map(|value| format!("{value}/{value}")),
462 any::<String>().prop_map(|value| format!("{value}\\{value}")),
463 any::<String>().prop_map(|value| format!("{value}.jsonl")),
464 any::<String>().prop_map(|value| format!("{value}é")),
465 ]
466 }
467
468 proptest! {
469 #[test]
470 fn validate_session_id_accepts_only_non_empty_safe_ascii_ids(id in valid_session_id_strategy()) {
471 let validated = validate_session_id(id.clone()).unwrap();
472
473 prop_assert_eq!(&validated, &id);
474 prop_assert!(!validated.is_empty());
475 prop_assert!(validated
476 .chars()
477 .all(|character| character.is_ascii_alphanumeric() || character == '_' || character == '-'));
478 }
479
480 #[test]
481 fn validate_session_id_rejects_generated_unsafe_ids(id in invalid_session_id_strategy()) {
482 prop_assert!(validate_session_id(id).is_err());
483 }
484 }
485
486 #[test]
487 fn recent_and_latest_session_reads_are_bounded_streaming_paths() {
488 let temp = TempDir::new().unwrap();
489 let manager = SessionManager::new(temp.path().join("sessions"));
490 let session = manager.create().unwrap();
491 for index in 0..25 {
492 session
493 .append(&SessionEvent::new(
494 "event",
495 session.id().to_string(),
496 temp.path().to_path_buf(),
497 json!({"index": index}),
498 ))
499 .unwrap();
500 }
501 let recent = session.read_recent_events(3, 4096).unwrap();
502 assert_eq!(recent.len(), 3);
503 assert_eq!(recent[0].payload["index"], 22);
504 assert!(session.latest_event_timestamp_bounded().unwrap().is_some());
505 assert_eq!(manager.most_recent().unwrap().unwrap().id(), session.id());
506 }
507
508 #[test]
509 fn read_recent_events_uses_original_jsonl_line_bytes() {
510 let temp = TempDir::new().unwrap();
511 let manager = SessionManager::new(temp.path().join("sessions"));
512 let session = manager.create().unwrap();
513 let padded = SessionEvent::new(
514 "event",
515 session.id().to_string(),
516 temp.path().to_path_buf(),
517 json!({"index": 1}),
518 );
519 let recent = SessionEvent::new(
520 "event",
521 session.id().to_string(),
522 temp.path().to_path_buf(),
523 json!({"index": 2}),
524 );
525 let padded_compact_len = serde_json::to_vec(&padded).unwrap().len();
526 let mut padded_line = serde_json::to_value(&padded).unwrap();
527 padded_line["padding"] = json!("x".repeat(2048));
528 let padded_line = serde_json::to_string(&padded_line).unwrap();
529 let recent_line = serde_json::to_string(&recent).unwrap();
530 fs::create_dir_all(session.path().parent().unwrap()).unwrap();
531 fs::write(session.path(), format!("{padded_line}\n{recent_line}\n")).unwrap();
532 crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
533
534 let max_bytes = padded_compact_len + recent_line.len() + 2;
535 let events = session.read_recent_events(10, max_bytes).unwrap();
536
537 assert_eq!(events.len(), 1);
538 assert_eq!(events[0].payload["index"], 2);
539 }
540
541 #[test]
542 fn tolerant_read_reports_malformed_lines_without_payload_leak() {
543 let temp = TempDir::new().unwrap();
544 let manager = SessionManager::new(temp.path().join("sessions"));
545 let session = manager.open("safe").unwrap();
546 let first = SessionEvent::new(
547 "user_input",
548 session.id().to_string(),
549 temp.path().to_path_buf(),
550 json!({"text":"before"}),
551 );
552 let second = SessionEvent::new(
553 "assistant_output",
554 session.id().to_string(),
555 temp.path().to_path_buf(),
556 json!({"text":"after"}),
557 );
558 fs::create_dir_all(session.path().parent().unwrap()).unwrap();
559 fs::write(
560 session.path(),
561 format!(
562 "{}\n{{\"access_token\":\"secret-token\",\n{}\n",
563 serde_json::to_string(&first).unwrap(),
564 serde_json::to_string(&second).unwrap()
565 ),
566 )
567 .unwrap();
568 crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
569
570 assert!(session.read_events().is_err());
571 let tolerant = session.read_events_tolerant().unwrap();
572
573 assert_eq!(tolerant.events.len(), 2);
574 assert_eq!(tolerant.events[0].payload["text"], "before");
575 assert_eq!(tolerant.events[1].payload["text"], "after");
576 assert_eq!(tolerant.diagnostics.len(), 1);
577 assert_eq!(tolerant.diagnostics[0].line, 2);
578 assert!(tolerant.diagnostics[0].message.contains("line 2"));
579 assert!(!tolerant.diagnostics[0].message.contains("secret-token"));
580 }
581
582 #[test]
583 fn tolerant_session_read_caps_malformed_diagnostics() {
584 let temp = TempDir::new().unwrap();
585 let manager = SessionManager::new(temp.path().join("sessions"));
586 let session = manager.open("safe").unwrap();
587 fs::create_dir_all(session.path().parent().unwrap()).unwrap();
588 let mut lines = Vec::new();
589 for index in 0..(MAX_TOLERANT_READ_DIAGNOSTICS + 10) {
590 lines.push(format!("not json {index}"));
591 }
592 fs::write(session.path(), lines.join("\n")).unwrap();
593 crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
594
595 let tolerant = session.read_events_tolerant().unwrap();
596
597 assert!(tolerant.events.is_empty());
598 assert_eq!(tolerant.diagnostics.len(), MAX_TOLERANT_READ_DIAGNOSTICS);
599 let summary = tolerant.diagnostics.last().unwrap();
600 assert_eq!(summary.line, 0);
601 assert!(summary.message.contains("omitted 10 additional"));
602 assert!(summary.message.contains("cap of 64"));
603 assert!(
604 !tolerant
605 .diagnostics
606 .iter()
607 .any(|diagnostic| diagnostic.message.contains("not json"))
608 );
609 }
610
611 #[test]
612 fn read_recent_events_tolerant_preserves_valid_events_after_malformed_line() {
613 let temp = TempDir::new().unwrap();
614 let manager = SessionManager::new(temp.path().join("sessions"));
615 let session = manager.open("safe").unwrap();
616 let old = SessionEvent::new(
617 "user_input",
618 session.id().to_string(),
619 temp.path().to_path_buf(),
620 json!({"text":"old"}),
621 );
622 let recent = SessionEvent::new(
623 "assistant_output",
624 session.id().to_string(),
625 temp.path().to_path_buf(),
626 json!({"text":"recent"}),
627 );
628 fs::create_dir_all(session.path().parent().unwrap()).unwrap();
629 fs::write(
630 session.path(),
631 format!(
632 "{}\nnot json\n{}\n",
633 serde_json::to_string(&old).unwrap(),
634 serde_json::to_string(&recent).unwrap()
635 ),
636 )
637 .unwrap();
638 crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
639
640 let tolerant = session.read_recent_events_tolerant(10, 4096).unwrap();
641
642 assert_eq!(tolerant.events.len(), 2);
643 assert_eq!(tolerant.events[1].payload["text"], "recent");
644 assert_eq!(tolerant.diagnostics[0].line, 2);
645 assert!(!tolerant.diagnostics[0].message.contains("not json"));
646 }
647
648 #[test]
649 fn read_recent_events_tolerant_bounds_malformed_input_bytes() {
650 let temp = TempDir::new().unwrap();
651 let manager = SessionManager::new(temp.path().join("sessions"));
652 let session = manager.open("safe").unwrap();
653 let valid = SessionEvent::new(
654 "user_input",
655 session.id().to_string(),
656 temp.path().to_path_buf(),
657 json!({"text":"valid"}),
658 );
659 fs::create_dir_all(session.path().parent().unwrap()).unwrap();
660 let malformed = "{".repeat(128);
661 fs::write(
662 session.path(),
663 format!("{}\n{malformed}\n", serde_json::to_string(&valid).unwrap()),
664 )
665 .unwrap();
666 crate::sessions::store::secure_test_session_root(session.path().parent().unwrap());
667
668 let error = session
669 .read_recent_events_tolerant(10, 64)
670 .unwrap_err()
671 .to_string();
672
673 assert!(
674 error.contains("tolerant recent read limit exceeded"),
675 "{error}"
676 );
677 assert!(error.contains("malformed bytes"), "{error}");
678 }
679}