Run one checkpoint cycle: dispatch checkpoint to all cores, collect LSNs, write checkpoint record, archive eligible WAL segments to cold storage (if configured), then truncate the WAL. Returns the global checkpoint LSN (min across all cores), or `None` if the checkpoint could not be completed (e.g., a core didn't respond).
(
dispatcher: &std::sync::Mutex<Dispatcher>,
tracker: &RequestTracker,
wal: &WalManager,
num_cores: usize,
timeout: Duration,
cold_storage: Option<std::sync::Arc<crate::storage
| 65 | /// Returns the global checkpoint LSN (min across all cores), or `None` if |
| 66 | /// the checkpoint could not be completed (e.g., a core didn't respond). |
| 67 | pub async fn run_checkpoint_cycle( |
| 68 | dispatcher: &std::sync::Mutex<Dispatcher>, |
| 69 | tracker: &RequestTracker, |
| 70 | wal: &WalManager, |
| 71 | num_cores: usize, |
| 72 | timeout: Duration, |
| 73 | cold_storage: Option<std::sync::Arc<crate::storage::cold::ColdStorage>>, |
| 74 | catalog: Option<&crate::control::security::catalog::SystemCatalog>, |
| 75 | ) -> Option<Lsn> { |
| 76 | if num_cores == 0 { |
| 77 | return None; |
| 78 | } |
| 79 | |
| 80 | // 1. Dispatch checkpoint requests to all cores. |
| 81 | let mut receivers = Vec::with_capacity(num_cores); |
| 82 | |
| 83 | { |
| 84 | let mut disp = dispatcher.lock().unwrap_or_else(|p| p.into_inner()); |
| 85 | |
| 86 | for core_id in 0..num_cores { |
| 87 | let request_id = |
| 88 | RequestId::new(CHECKPOINT_REQUEST_COUNTER.fetch_add(1, Ordering::Relaxed)); |
| 89 | let vshard_id = VShardId::new(core_id as u32); |
| 90 | |
| 91 | let request = Request { |
| 92 | request_id, |
| 93 | tenant_id: TenantId::new(0), // System-level checkpoint. |
| 94 | database_id: DatabaseId::DEFAULT, |
| 95 | vshard_id, |
| 96 | plan: PhysicalPlan::Meta(MetaOp::Checkpoint), |
| 97 | deadline: std::time::Instant::now() + timeout, |
| 98 | priority: Priority::Background, |
| 99 | trace_id: TraceId::generate(), |
| 100 | consistency: ReadConsistency::Eventual, |
| 101 | idempotency_key: None, |
| 102 | event_source: crate::event::EventSource::User, |
| 103 | user_roles: Vec::new(), |
| 104 | user_id: None, |
| 105 | statement_digest: None, |
| 106 | }; |
| 107 | |
| 108 | let rx = tracker.register(request_id); |
| 109 | |
| 110 | if let Err(e) = disp.dispatch_to_core(core_id, request) { |
| 111 | warn!( |
| 112 | core_id, |
| 113 | error = %e, |
| 114 | "failed to dispatch checkpoint to core" |
| 115 | ); |
| 116 | tracker.cancel(&request_id); |
| 117 | continue; |
| 118 | } |
| 119 | |
| 120 | receivers.push((core_id, request_id, rx)); |
| 121 | } |
| 122 | } |
| 123 | |
| 124 | if receivers.is_empty() { |
no test coverage detected