| 1 | use std::{ |
| 2 | collections::HashSet, |
| 3 | sync::{ |
| 4 | Arc, |
| 5 | atomic::{AtomicBool, Ordering}, |
| 6 | }, |
| 7 | time::Duration, |
| 8 | }; |
| 9 | |
| 10 | use matrix_sdk::{Room, deserialized_responses::TimelineEvent, room::MessagesOptions}; |
| 11 | use matrix_sdk_ui::sync_service::SyncService; |
| 12 | use ruma::{OwnedEventId, time::Instant}; |
| 13 | use tokio::sync::{OwnedMutexGuard, mpsc, oneshot, watch}; |
| 14 | |
| 15 | use super::{ |
| 16 | executor::{JoinHandle, spawn}, |
| 17 | timeline::SessionContext, |
| 18 | }; |
| 19 | use crate::{ |
| 20 | CommandOutput, Error, ErrorKind, Event, HistoryContinuity, HistoryRepairBudget, |
| 21 | HistoryRepairCommand, HistoryRepairPhase, HistoryRepairPolicy, HistoryRepairReport, |
| 22 | HistoryRepairStop, HistoryRepairStrategy, transport::Responder, |
| 23 | }; |
| 24 | |
| 25 | const PAGE_SIZE: u16 = 50; |
| 26 | const STAGING_LIMIT: usize = 10_000; |
| 27 | |
| 28 | pub(super) struct InstallRequest { |
| 29 | pub(super) events: Vec<TimelineEvent>, |
| 30 | pub(super) prev_batch: Option<String>, |
| 31 | pub(super) reply: oneshot::Sender<Result<(), Error>>, |
| 32 | } |
| 33 | |
| 34 | struct Stage { |
| 35 | events: Vec<TimelineEvent>, |
| 36 | cursor: Option<String>, |
| 37 | anchor: Option<OwnedEventId>, |
| 38 | report: HistoryRepairReport, |
| 39 | } |
| 40 | |
| 41 | struct Running { |
| 42 | task: JoinHandle<Stage>, |
| 43 | cancel: watch::Sender<bool>, |
| 44 | reply: Responder, |
| 45 | } |
| 46 | |
| 47 | pub(super) struct RepairController { |
| 48 | room: Room, |
| 49 | context: SessionContext, |
| 50 | installs: mpsc::Sender<InstallRequest>, |
| 51 | requests: mpsc::Receiver<InstallRequest>, |
| 52 | running: Option<Running>, |
| 53 | stage: Option<Stage>, |
| 54 | } |
| 55 | |
| 56 | impl RepairController { |
| 57 | pub(super) fn new(room: Room, context: SessionContext) -> Self { |
| 58 | let (installs, requests) = mpsc::channel(1); |
| 59 | Self { |
| 60 | room, |
| 61 | context, |
| 62 | installs, |
| 63 | requests, |
| 64 | running: None, |
| 65 | stage: None, |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | pub(super) fn is_running(&self) -> bool { |
| 70 | self.running.is_some() |
| 71 | } |
| 72 | |
| 73 | pub(super) async fn execute(&mut self, command: HistoryRepairCommand, reply: Responder) { |
| 74 | if matches!(command, HistoryRepairCommand::Cancel { .. }) { |
| 75 | if let Some(running) = &self.running { |
| 76 | running.cancel.send_replace(true); |
| 77 | } |
| 78 | reply.done(Ok(())); |
| 79 | return; |
| 80 | } |
| 81 | if self.is_running() { |
| 82 | reply.done(Err(Error::new( |
| 83 | ErrorKind::RoomBusy, |
| 84 | "room history repair is running", |
| 85 | ))); |
| 86 | return; |
| 87 | } |
| 88 | let (stage, budget) = match command { |
| 89 | HistoryRepairCommand::Start { policy, budget, .. } => { |
| 90 | if let Err(error) = validate_budget(&budget) { |
| 91 | reply.done(Err(error)); |
| 92 | return; |
| 93 | } |
| 94 | let policy = match policy { |
| 95 | Some(policy) => policy, |
| 96 | None => match self.context.mentions.repair_policy().await { |
| 97 | Ok(policy) => policy, |
| 98 | Err(error) => { |
| 99 | reply.done(Err(error)); |
| 100 | return; |
| 101 | } |
| 102 | }, |
| 103 | }; |
| 104 | let coverage = match self |
| 105 | .context |
| 106 | .mentions |
| 107 | .repair_coverage(self.room.room_id().to_owned()) |
| 108 | .await |
| 109 | { |
| 110 | Ok(coverage) => coverage, |
| 111 | Err(error) => { |
| 112 | reply.done(Err(error)); |
| 113 | return; |
| 114 | } |
| 115 | }; |
| 116 | let duration = coverage.duration_ms(); |
| 117 | let report = HistoryRepairReport { |
| 118 | room_id: self.room.room_id().to_owned(), |
| 119 | requested_policy: policy.clone(), |
| 120 | strategy: HistoryRepairStrategy::Recent, |
| 121 | coverage_duration_ms: duration, |
| 122 | threshold_evaluated: duration.is_some() |
| 123 | && coverage.unresolved |
| 124 | && matches!(policy, HistoryRepairPolicy::BridgeShortGaps { .. }), |
| 125 | pages: 0, |
| 126 | events: 0, |
| 127 | phase: HistoryRepairPhase::FetchingRecent, |
| 128 | continuity: HistoryContinuity::Unresolved, |
| 129 | recent_available: false, |
| 130 | stop: None, |
| 131 | failure: None, |
| 132 | }; |
| 133 | ( |
| 134 | Stage { |
| 135 | events: Vec::new(), |
| 136 | cursor: None, |
| 137 | anchor: None, |
| 138 | report, |
| 139 | }, |
| 140 | budget, |
| 141 | ) |
| 142 | } |
| 143 | HistoryRepairCommand::Continue { budget, .. } => { |
| 144 | if let Err(error) = validate_budget(&budget) { |
| 145 | reply.done(Err(error)); |
| 146 | return; |
| 147 | } |
| 148 | let Some(mut stage) = self.stage.take() else { |
| 149 | reply.done(Err(Error::new( |
| 150 | ErrorKind::Invalid, |
| 151 | "no history bridge to continue", |
| 152 | ))); |
| 153 | return; |
| 154 | }; |
| 155 | if !stage.report.recent_available |
| 156 | || stage.report.continuity != HistoryContinuity::Unresolved |
| 157 | { |
| 158 | reply.done(Err(Error::new( |
| 159 | ErrorKind::Invalid, |
| 160 | "history bridge is not resumable", |
| 161 | ))); |
| 162 | self.stage = Some(stage); |
| 163 | return; |
| 164 | } |
| 165 | stage.report.strategy = HistoryRepairStrategy::Bridge; |
| 166 | stage.report.phase = HistoryRepairPhase::Bridging; |
| 167 | stage.report.stop = None; |
| 168 | stage.report.failure = None; |
| 169 | (stage, budget) |
| 170 | } |
| 171 | HistoryRepairCommand::StartRecent { .. } => { |
| 172 | let Some(mut stage) = self.stage.take() else { |
| 173 | reply.done(Err(Error::new( |
| 174 | ErrorKind::Invalid, |
| 175 | "no recent history repair", |
| 176 | ))); |
| 177 | return; |
| 178 | }; |
| 179 | if !stage.report.recent_available { |
| 180 | self.stage = Some(stage); |
| 181 | reply.done(Err(Error::new( |
| 182 | ErrorKind::Invalid, |
| 183 | "recent history is not available, retry the repair", |
| 184 | ))); |
| 185 | return; |
| 186 | } |
| 187 | stage.report.strategy = HistoryRepairStrategy::Recent; |
| 188 | stage.report.phase = HistoryRepairPhase::Ready; |
| 189 | stage.report.stop = None; |
| 190 | stage.report.failure = None; |
| 191 | let report = stage.report.clone(); |
| 192 | let published = self |
| 193 | .context |
| 194 | .events |
| 195 | .publish_repair(report.clone()) |
| 196 | .await |
| 197 | .map_err(|_| Error::interrupted()); |
| 198 | reply.send(published.map(|_| CommandOutput::HistoryRepair(report))); |
| 199 | return; |
| 200 | } |
| 201 | _ => { |
| 202 | reply.done(Err(Error::new( |
| 203 | ErrorKind::Invalid, |
| 204 | "repair command was misrouted", |
| 205 | ))); |
| 206 | return; |
| 207 | } |
| 208 | }; |
| 209 | let (cancel, receiver) = watch::channel(false); |
| 210 | let room = self.room.clone(); |
| 211 | let context = self.context.clone(); |
| 212 | let installs = self.installs.clone(); |
| 213 | let task = |
| 214 | spawn(async move { run(room, context, installs, receiver, stage, budget).await }); |
| 215 | self.running = Some(Running { |
| 216 | task, |
| 217 | cancel, |
| 218 | reply, |
| 219 | }); |
| 220 | } |
| 221 | |
| 222 | pub(super) async fn next_install(&mut self) -> Option<InstallRequest> { |
| 223 | let running = self.running.as_mut()?; |
| 224 | tokio::select! { |
| 225 | request = self.requests.recv() => request, |
| 226 | result = &mut running.task => { |
| 227 | let running = self.running.take().expect("repair task exists"); |
| 228 | match result { |
| 229 | Ok(stage) => { |
| 230 | let report = stage.report.clone(); |
| 231 | let published = self.context.events.publish_repair(report.clone()).await |
| 232 | .map_err(|_| Error::interrupted()); |
| 233 | running.reply.send(published.map(|_| CommandOutput::HistoryRepair(report))); |
| 234 | self.stage = Some(stage); |
| 235 | } |
| 236 | Err(_) => running.reply.done(Err(Error::interrupted())), |
| 237 | } |
| 238 | None |
| 239 | } |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | pub(super) async fn shutdown(&mut self) { |
| 244 | if let Some(mut running) = self.running.take() { |
| 245 | running.task.abort(); |
| 246 | match (&mut running.task).await { |
| 247 | Ok(stage) => { |
| 248 | let report = stage.report; |
| 249 | let published = self |
| 250 | .context |
| 251 | .events |
| 252 | .publish_repair(report.clone()) |
| 253 | .await |
| 254 | .map_err(|_| Error::interrupted()); |
| 255 | running |
| 256 | .reply |
| 257 | .send(published.map(|_| CommandOutput::HistoryRepair(report))); |
| 258 | } |
| 259 | Err(_) => running.reply.done(Err(Error::interrupted())), |
| 260 | } |
| 261 | } |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | impl Drop for RepairController { |
| 266 | fn drop(&mut self) { |
| 267 | if let Some(running) = &self.running { |
| 268 | running.task.abort(); |
| 269 | } |
| 270 | } |
| 271 | } |
| 272 | |
| 273 | fn validate_budget(budget: &HistoryRepairBudget) -> Result<(), Error> { |
| 274 | if !(1..=100).contains(&budget.max_pages) |
| 275 | || !(50..=10_000).contains(&budget.max_events) |
| 276 | || !(1..=60_000).contains(&budget.max_duration_ms) |
| 277 | { |
| 278 | return Err(Error::new( |
| 279 | ErrorKind::Invalid, |
| 280 | "repair budget requires 1 to 100 pages, 50 to 10000 events, and 1 to 60000 milliseconds", |
| 281 | )); |
| 282 | } |
| 283 | Ok(()) |
| 284 | } |
| 285 | |
| 286 | fn should_bridge(policy: &HistoryRepairPolicy, duration: Option<u64>, boundary: bool) -> bool { |
| 287 | match policy { |
| 288 | HistoryRepairPolicy::Bridge => true, |
| 289 | HistoryRepairPolicy::StartRecent => false, |
| 290 | HistoryRepairPolicy::BridgeShortGaps { threshold_ms } => { |
| 291 | boundary |
| 292 | && *threshold_ms > 0 |
| 293 | && duration.is_some_and(|duration| duration <= *threshold_ms) |
| 294 | } |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | async fn run( |
| 299 | room: Room, |
| 300 | context: SessionContext, |
| 301 | installs: mpsc::Sender<InstallRequest>, |
| 302 | mut cancel: watch::Receiver<bool>, |
| 303 | mut stage: Stage, |
| 304 | budget: HistoryRepairBudget, |
| 305 | ) -> Stage { |
| 306 | let started = Instant::now(); |
| 307 | let initial_pages = stage.report.pages; |
| 308 | let initial_events = stage.report.events; |
| 309 | let result = async { |
| 310 | if !stage.report.recent_available { |
| 311 | let (cache, _handles) = room.event_cache().await.map_err(cache_error)?; |
| 312 | stage.anchor = cache |
| 313 | .events() |
| 314 | .await |
| 315 | .map_err(cache_error)? |
| 316 | .iter() |
| 317 | .find_map(TimelineEvent::event_id); |
| 318 | let _ = context |
| 319 | .events |
| 320 | .send(Event::HistoryRepairChanged(stage.report.clone())) |
| 321 | .await; |
| 322 | let guard = pause_sync(&context, &mut cancel, &started, &budget).await?; |
| 323 | let page = fetch(&room, None, &mut cancel, &started, &budget).await?; |
| 324 | stage.report.pages += 1; |
| 325 | stage.report.events += page.chunk.len() as u32; |
| 326 | stage.cursor = page.end; |
| 327 | stage.events = page.chunk.into_iter().rev().collect(); |
| 328 | deduplicate(&mut stage.events); |
| 329 | install(&installs, &stage).await?; |
| 330 | stage.report.recent_available = true; |
| 331 | if contains_anchor(&stage) { |
| 332 | stage.report.strategy = HistoryRepairStrategy::Overlap; |
| 333 | stage.report.continuity = HistoryContinuity::Verified; |
| 334 | } else if stage.cursor.is_none() { |
| 335 | stage.report.continuity = HistoryContinuity::AccessibleStart; |
| 336 | } |
| 337 | if stage.report.continuity != HistoryContinuity::Unresolved { |
| 338 | context |
| 339 | .mentions |
| 340 | .repair_resolved(room.room_id().to_owned()) |
| 341 | .await?; |
| 342 | } |
| 343 | guard.finish().await; |
| 344 | let _ = context |
| 345 | .events |
| 346 | .send(Event::HistoryRepairChanged(stage.report.clone())) |
| 347 | .await; |
| 348 | if stage.report.continuity != HistoryContinuity::Unresolved { |
| 349 | return Ok(()); |
| 350 | } |
| 351 | if !should_bridge( |
| 352 | &stage.report.requested_policy, |
| 353 | stage.report.coverage_duration_ms, |
| 354 | stage.report.threshold_evaluated, |
| 355 | ) { |
| 356 | return Ok(()); |
| 357 | } |
| 358 | stage.report.strategy = HistoryRepairStrategy::Bridge; |
| 359 | } |
| 360 | stage.report.phase = HistoryRepairPhase::Bridging; |
| 361 | let _ = context |
| 362 | .events |
| 363 | .send(Event::HistoryRepairChanged(stage.report.clone())) |
| 364 | .await; |
| 365 | if contains_anchor(&stage) { |
| 366 | stage.report.continuity = HistoryContinuity::Verified; |
| 367 | } else if stage.cursor.is_none() { |
| 368 | stage.report.continuity = HistoryContinuity::AccessibleStart; |
| 369 | } |
| 370 | while stage.report.continuity == HistoryContinuity::Unresolved { |
| 371 | if *cancel.borrow() { |
| 372 | stage.report.stop = Some(HistoryRepairStop::Cancelled); |
| 373 | break; |
| 374 | } |
| 375 | if stage.report.pages - initial_pages + 1 >= u32::from(budget.max_pages) |
| 376 | || stage.report.events - initial_events + 2 * u32::from(PAGE_SIZE) |
| 377 | > budget.max_events |
| 378 | || stage.events.len() + 2 * usize::from(PAGE_SIZE) > STAGING_LIMIT |
| 379 | || started.elapsed() >= Duration::from_millis(budget.max_duration_ms) |
| 380 | { |
| 381 | stage.report.stop = Some(HistoryRepairStop::Budget); |
| 382 | break; |
| 383 | } |
| 384 | if stage.anchor.is_none() { |
| 385 | stage.report.stop = Some(HistoryRepairStop::NoAnchor); |
| 386 | break; |
| 387 | } |
| 388 | let Some(cursor) = stage.cursor.clone() else { |
| 389 | break; |
| 390 | }; |
| 391 | let page = fetch(&room, Some(&cursor), &mut cancel, &started, &budget).await?; |
| 392 | stage.report.pages += 1; |
| 393 | stage.report.events += page.chunk.len() as u32; |
| 394 | if page.end.as_deref() == Some(&cursor) { |
| 395 | stage.report.stop = Some(HistoryRepairStop::CursorStalled); |
| 396 | break; |
| 397 | } |
| 398 | let mut older: Vec<_> = page.chunk.into_iter().rev().collect(); |
| 399 | older.append(&mut stage.events); |
| 400 | deduplicate(&mut older); |
| 401 | stage.events = older; |
| 402 | stage.cursor = page.end; |
| 403 | if contains_anchor(&stage) { |
| 404 | stage.report.continuity = HistoryContinuity::Verified; |
| 405 | } else if stage.cursor.is_none() { |
| 406 | stage.report.continuity = HistoryContinuity::AccessibleStart; |
| 407 | } |
| 408 | if stage.report.continuity == HistoryContinuity::Unresolved { |
| 409 | let _ = context |
| 410 | .events |
| 411 | .send(Event::HistoryRepairChanged(stage.report.clone())) |
| 412 | .await; |
| 413 | } |
| 414 | } |
| 415 | if stage.report.continuity != HistoryContinuity::Unresolved |
| 416 | && stage.report.strategy == HistoryRepairStrategy::Bridge |
| 417 | { |
| 418 | let guard = pause_sync(&context, &mut cancel, &started, &budget).await?; |
| 419 | let head = fetch(&room, None, &mut cancel, &started, &budget).await?; |
| 420 | stage.report.pages += 1; |
| 421 | stage.report.events += head.chunk.len() as u32; |
| 422 | let head: Vec<_> = head.chunk.into_iter().rev().collect(); |
| 423 | if !extend_head(&mut stage.events, head) { |
| 424 | stage.report.continuity = HistoryContinuity::Unresolved; |
| 425 | stage.report.stop = Some(HistoryRepairStop::HistoryUnavailable); |
| 426 | return Ok(()); |
| 427 | } |
| 428 | install(&installs, &stage).await?; |
| 429 | context |
| 430 | .mentions |
| 431 | .repair_resolved(room.room_id().to_owned()) |
| 432 | .await?; |
| 433 | guard.finish().await; |
| 434 | } |
| 435 | Ok::<_, Error>(()) |
| 436 | } |
| 437 | .await; |
| 438 | if let Err(error) = result { |
| 439 | if error.kind == ErrorKind::Interrupted { |
| 440 | stage.report.stop = Some(HistoryRepairStop::Cancelled); |
| 441 | } else if error.kind == ErrorKind::RoomBusy { |
| 442 | stage.report.stop = Some(HistoryRepairStop::Budget); |
| 443 | } else { |
| 444 | stage.report.failure = Some(error); |
| 445 | } |
| 446 | stage.report.continuity = HistoryContinuity::Unresolved; |
| 447 | } |
| 448 | stage.report.phase = if stage.report.stop.is_some() || stage.report.failure.is_some() { |
| 449 | HistoryRepairPhase::Paused |
| 450 | } else { |
| 451 | HistoryRepairPhase::Ready |
| 452 | }; |
| 453 | stage |
| 454 | } |
| 455 | |
| 456 | fn contains_anchor(stage: &Stage) -> bool { |
| 457 | stage.anchor.as_ref().is_some_and(|anchor| { |
| 458 | stage |
| 459 | .events |
| 460 | .iter() |
| 461 | .any(|event| event.event_id().as_ref() == Some(anchor)) |
| 462 | }) |
| 463 | } |
| 464 | |
| 465 | fn deduplicate(events: &mut Vec<TimelineEvent>) { |
| 466 | let mut seen = HashSet::new(); |
| 467 | events.retain(|event| event.event_id().is_some_and(|id| seen.insert(id))); |
| 468 | } |
| 469 | |
| 470 | fn extend_head(events: &mut Vec<TimelineEvent>, head: Vec<TimelineEvent>) -> bool { |
| 471 | let ids: HashSet<_> = events.iter().filter_map(TimelineEvent::event_id).collect(); |
| 472 | let Some(overlap) = head |
| 473 | .iter() |
| 474 | .position(|event| event.event_id().is_some_and(|id| ids.contains(&id))) |
| 475 | else { |
| 476 | return events.is_empty() && head.is_empty(); |
| 477 | }; |
| 478 | events.extend(head.into_iter().skip(overlap + 1)); |
| 479 | deduplicate(events); |
| 480 | true |
| 481 | } |
| 482 | |
| 483 | async fn install(installs: &mpsc::Sender<InstallRequest>, stage: &Stage) -> Result<(), Error> { |
| 484 | let (reply, response) = oneshot::channel(); |
| 485 | installs |
| 486 | .send(InstallRequest { |
| 487 | events: stage.events.clone(), |
| 488 | prev_batch: stage.cursor.clone(), |
| 489 | reply, |
| 490 | }) |
| 491 | .await |
| 492 | .map_err(|_| Error::interrupted())?; |
| 493 | response.await.map_err(|_| Error::interrupted())? |
| 494 | } |
| 495 | |
| 496 | async fn fetch( |
| 497 | room: &Room, |
| 498 | from: Option<&str>, |
| 499 | cancel: &mut watch::Receiver<bool>, |
| 500 | started: &Instant, |
| 501 | budget: &HistoryRepairBudget, |
| 502 | ) -> Result<matrix_sdk::room::Messages, Error> { |
| 503 | let remaining = Duration::from_millis(budget.max_duration_ms).saturating_sub(started.elapsed()); |
| 504 | if *cancel.borrow() { |
| 505 | return Err(Error::interrupted()); |
| 506 | } |
| 507 | let mut options = MessagesOptions::backward().from(from); |
| 508 | options.limit = PAGE_SIZE.into(); |
| 509 | tokio::select! { |
| 510 | _ = cancel.changed() => Err(Error::interrupted()), |
| 511 | result = matrix_sdk::timeout::timeout(room.messages(options), remaining) => { |
| 512 | let page = result.map_err(|_| Error::new(ErrorKind::RoomBusy, "repair work duration exhausted"))? |
| 513 | .map_err(|source| Error::from_sdk("fetching repair history failed", source))?; |
| 514 | if page.chunk.len() > usize::from(PAGE_SIZE) { |
| 515 | return Err(Error::new(ErrorKind::Invalid, "server exceeded the repair page limit")); |
| 516 | } |
| 517 | Ok(page) |
| 518 | } |
| 519 | } |
| 520 | } |
| 521 | |
| 522 | struct SyncPause { |
| 523 | sync: Arc<SyncService>, |
| 524 | shutdown: Arc<AtomicBool>, |
| 525 | gate: Option<OwnedMutexGuard<()>>, |
| 526 | } |
| 527 | |
| 528 | impl SyncPause { |
| 529 | async fn finish(mut self) { |
| 530 | if !self.shutdown.load(Ordering::Acquire) { |
| 531 | self.sync.start().await; |
| 532 | } |
| 533 | self.gate.take(); |
| 534 | } |
| 535 | } |
| 536 | |
| 537 | impl Drop for SyncPause { |
| 538 | fn drop(&mut self) { |
| 539 | if let Some(gate) = self.gate.take() { |
| 540 | let sync = self.sync.clone(); |
| 541 | let shutdown = self.shutdown.clone(); |
| 542 | spawn(async move { |
| 543 | if !shutdown.load(Ordering::Acquire) { |
| 544 | sync.start().await; |
| 545 | } |
| 546 | drop(gate); |
| 547 | }); |
| 548 | } |
| 549 | } |
| 550 | } |
| 551 | |
| 552 | async fn pause_sync( |
| 553 | context: &SessionContext, |
| 554 | cancel: &mut watch::Receiver<bool>, |
| 555 | started: &Instant, |
| 556 | budget: &HistoryRepairBudget, |
| 557 | ) -> Result<SyncPause, Error> { |
| 558 | let remaining = Duration::from_millis(budget.max_duration_ms).saturating_sub(started.elapsed()); |
| 559 | let gate = tokio::select! { |
| 560 | _ = cancel.changed() => return Err(Error::interrupted()), |
| 561 | result = matrix_sdk::timeout::timeout(context.repair_gate.clone().lock_owned(), remaining) => |
| 562 | result.map_err(|_| Error::new(ErrorKind::RoomBusy, "repair work duration exhausted"))?, |
| 563 | }; |
| 564 | if context.repair_shutdown.load(Ordering::Acquire) { |
| 565 | return Err(Error::interrupted()); |
| 566 | } |
| 567 | let guard = SyncPause { |
| 568 | sync: context.sync.clone(), |
| 569 | shutdown: context.repair_shutdown.clone(), |
| 570 | gate: Some(gate), |
| 571 | }; |
| 572 | context.sync.stop().await; |
| 573 | Ok(guard) |
| 574 | } |
| 575 | |
| 576 | fn cache_error(source: matrix_sdk::event_cache::EventCacheError) -> Error { |
| 577 | Error::with_source( |
| 578 | ErrorKind::Store, |
| 579 | "room history cache operation failed", |
| 580 | source, |
| 581 | ) |
| 582 | } |
| 583 | |
| 584 | #[cfg(test)] |
| 585 | mod tests { |
| 586 | use super::*; |
| 587 | |
| 588 | #[test] |
| 589 | fn threshold_requires_known_coverage_and_an_independent_boundary() { |
| 590 | let policy = HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 100 }; |
| 591 | assert!(should_bridge(&policy, Some(100), true)); |
| 592 | assert!(should_bridge(&policy, Some(99), true)); |
| 593 | assert!(!should_bridge(&policy, Some(101), true)); |
| 594 | assert!(!should_bridge(&policy, None, true)); |
| 595 | assert!(!should_bridge(&policy, Some(1), false)); |
| 596 | assert!(!should_bridge( |
| 597 | &HistoryRepairPolicy::BridgeShortGaps { threshold_ms: 0 }, |
| 598 | Some(0), |
| 599 | true |
| 600 | )); |
| 601 | assert!(should_bridge(&HistoryRepairPolicy::Bridge, None, false)); |
| 602 | assert!(!should_bridge( |
| 603 | &HistoryRepairPolicy::StartRecent, |
| 604 | Some(1), |
| 605 | true |
| 606 | )); |
| 607 | } |
| 608 | } |