LCOV - code coverage report
Current view: top level - safekeeper/src - timeline.rs (source / functions) Coverage Total Hit
Test: 4f58e98c51285c7fa348e0b410c88a10caf68ad2.info Lines: 0.0 % 689 0
Test Date: 2025-01-07 20:58:07 Functions: 0.0 % 113 0

            Line data    Source code
       1              : //! This module implements Timeline lifecycle management and has all necessary code
       2              : //! to glue together SafeKeeper and all other background services.
       3              : 
       4              : use anyhow::{anyhow, bail, Result};
       5              : use camino::{Utf8Path, Utf8PathBuf};
       6              : use remote_storage::RemotePath;
       7              : use safekeeper_api::models::{PeerInfo, TimelineTermBumpResponse};
       8              : use safekeeper_api::Term;
       9              : use tokio::fs::{self};
      10              : use tokio_util::sync::CancellationToken;
      11              : use utils::id::TenantId;
      12              : use utils::sync::gate::Gate;
      13              : 
      14              : use std::cmp::max;
      15              : use std::ops::{Deref, DerefMut};
      16              : use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
      17              : use std::sync::Arc;
      18              : use std::time::Duration;
      19              : use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
      20              : use tokio::{sync::watch, time::Instant};
      21              : use tracing::*;
      22              : use utils::http::error::ApiError;
      23              : use utils::{
      24              :     id::{NodeId, TenantTimelineId},
      25              :     lsn::Lsn,
      26              : };
      27              : 
      28              : use storage_broker::proto::SafekeeperTimelineInfo;
      29              : use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId;
      30              : 
      31              : use crate::control_file;
      32              : use crate::rate_limit::RateLimiter;
      33              : use crate::receive_wal::WalReceivers;
      34              : use crate::safekeeper::{AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, TermLsn};
      35              : use crate::send_wal::WalSenders;
      36              : use crate::state::{EvictionState, TimelineMemState, TimelinePersistentState, TimelineState};
      37              : use crate::timeline_guard::ResidenceGuard;
      38              : use crate::timeline_manager::{AtomicStatus, ManagerCtl};
      39              : use crate::timelines_set::TimelinesSet;
      40              : use crate::wal_backup::{self, remote_timeline_path};
      41              : use crate::wal_backup_partial::PartialRemoteSegment;
      42              : 
      43              : use crate::metrics::{FullTimelineInfo, WalStorageMetrics, MISC_OPERATION_SECONDS};
      44              : use crate::wal_storage::{Storage as wal_storage_iface, WalReader};
      45              : use crate::SafeKeeperConf;
      46              : use crate::{debug_dump, timeline_manager, wal_storage};
      47              : 
      48            0 : fn peer_info_from_sk_info(sk_info: &SafekeeperTimelineInfo, ts: Instant) -> PeerInfo {
      49            0 :     PeerInfo {
      50            0 :         sk_id: NodeId(sk_info.safekeeper_id),
      51            0 :         term: sk_info.term,
      52            0 :         last_log_term: sk_info.last_log_term,
      53            0 :         flush_lsn: Lsn(sk_info.flush_lsn),
      54            0 :         commit_lsn: Lsn(sk_info.commit_lsn),
      55            0 :         local_start_lsn: Lsn(sk_info.local_start_lsn),
      56            0 :         pg_connstr: sk_info.safekeeper_connstr.clone(),
      57            0 :         http_connstr: sk_info.http_connstr.clone(),
      58            0 :         ts,
      59            0 :     }
      60            0 : }
      61              : 
      62              : // vector-based node id -> peer state map with very limited functionality we
      63              : // need.
      64              : #[derive(Debug, Clone, Default)]
      65              : pub struct PeersInfo(pub Vec<PeerInfo>);
      66              : 
      67              : impl PeersInfo {
      68            0 :     fn get(&mut self, id: NodeId) -> Option<&mut PeerInfo> {
      69            0 :         self.0.iter_mut().find(|p| p.sk_id == id)
      70            0 :     }
      71              : 
      72            0 :     fn upsert(&mut self, p: &PeerInfo) {
      73            0 :         match self.get(p.sk_id) {
      74            0 :             Some(rp) => *rp = p.clone(),
      75            0 :             None => self.0.push(p.clone()),
      76              :         }
      77            0 :     }
      78              : }
      79              : 
      80              : pub type ReadGuardSharedState<'a> = RwLockReadGuard<'a, SharedState>;
      81              : 
      82              : /// WriteGuardSharedState is a wrapper around `RwLockWriteGuard<SharedState>` that
      83              : /// automatically updates `watch::Sender` channels with state on drop.
      84              : pub struct WriteGuardSharedState<'a> {
      85              :     tli: Arc<Timeline>,
      86              :     guard: RwLockWriteGuard<'a, SharedState>,
      87              : }
      88              : 
      89              : impl<'a> WriteGuardSharedState<'a> {
      90            0 :     fn new(tli: Arc<Timeline>, guard: RwLockWriteGuard<'a, SharedState>) -> Self {
      91            0 :         WriteGuardSharedState { tli, guard }
      92            0 :     }
      93              : }
      94              : 
      95              : impl Deref for WriteGuardSharedState<'_> {
      96              :     type Target = SharedState;
      97              : 
      98            0 :     fn deref(&self) -> &Self::Target {
      99            0 :         &self.guard
     100            0 :     }
     101              : }
     102              : 
     103              : impl DerefMut for WriteGuardSharedState<'_> {
     104            0 :     fn deref_mut(&mut self) -> &mut Self::Target {
     105            0 :         &mut self.guard
     106            0 :     }
     107              : }
     108              : 
     109              : impl Drop for WriteGuardSharedState<'_> {
     110            0 :     fn drop(&mut self) {
     111            0 :         let term_flush_lsn =
     112            0 :             TermLsn::from((self.guard.sk.last_log_term(), self.guard.sk.flush_lsn()));
     113            0 :         let commit_lsn = self.guard.sk.state().inmem.commit_lsn;
     114            0 : 
     115            0 :         let _ = self.tli.term_flush_lsn_watch_tx.send_if_modified(|old| {
     116            0 :             if *old != term_flush_lsn {
     117            0 :                 *old = term_flush_lsn;
     118            0 :                 true
     119              :             } else {
     120            0 :                 false
     121              :             }
     122            0 :         });
     123            0 : 
     124            0 :         let _ = self.tli.commit_lsn_watch_tx.send_if_modified(|old| {
     125            0 :             if *old != commit_lsn {
     126            0 :                 *old = commit_lsn;
     127            0 :                 true
     128              :             } else {
     129            0 :                 false
     130              :             }
     131            0 :         });
     132            0 : 
     133            0 :         // send notification about shared state update
     134            0 :         self.tli.shared_state_version_tx.send_modify(|old| {
     135            0 :             *old += 1;
     136            0 :         });
     137            0 :     }
     138              : }
     139              : 
     140              : /// This structure is stored in shared state and represents the state of the timeline.
     141              : ///
     142              : /// Usually it holds SafeKeeper, but it also supports offloaded timeline state. In this
     143              : /// case, SafeKeeper is not available (because WAL is not present on disk) and all
     144              : /// operations can be done only with control file.
     145              : pub enum StateSK {
     146              :     Loaded(SafeKeeper<control_file::FileStorage, wal_storage::PhysicalStorage>),
     147              :     Offloaded(Box<TimelineState<control_file::FileStorage>>),
     148              :     // Not used, required for moving between states.
     149              :     Empty,
     150              : }
     151              : 
     152              : impl StateSK {
     153            0 :     pub fn flush_lsn(&self) -> Lsn {
     154            0 :         match self {
     155            0 :             StateSK::Loaded(sk) => sk.wal_store.flush_lsn(),
     156            0 :             StateSK::Offloaded(state) => match state.eviction_state {
     157            0 :                 EvictionState::Offloaded(flush_lsn) => flush_lsn,
     158            0 :                 _ => panic!("StateSK::Offloaded mismatches with eviction_state from control_file"),
     159              :             },
     160            0 :             StateSK::Empty => unreachable!(),
     161              :         }
     162            0 :     }
     163              : 
     164              :     /// Get a reference to the control file's timeline state.
     165            0 :     pub fn state(&self) -> &TimelineState<control_file::FileStorage> {
     166            0 :         match self {
     167            0 :             StateSK::Loaded(sk) => &sk.state,
     168            0 :             StateSK::Offloaded(ref s) => s,
     169            0 :             StateSK::Empty => unreachable!(),
     170              :         }
     171            0 :     }
     172              : 
     173            0 :     pub fn state_mut(&mut self) -> &mut TimelineState<control_file::FileStorage> {
     174            0 :         match self {
     175            0 :             StateSK::Loaded(sk) => &mut sk.state,
     176            0 :             StateSK::Offloaded(ref mut s) => s,
     177            0 :             StateSK::Empty => unreachable!(),
     178              :         }
     179            0 :     }
     180              : 
     181            0 :     pub fn last_log_term(&self) -> Term {
     182            0 :         self.state()
     183            0 :             .acceptor_state
     184            0 :             .get_last_log_term(self.flush_lsn())
     185            0 :     }
     186              : 
     187            0 :     pub async fn term_bump(&mut self, to: Option<Term>) -> Result<TimelineTermBumpResponse> {
     188            0 :         self.state_mut().term_bump(to).await
     189            0 :     }
     190              : 
     191              :     /// Close open WAL files to release FDs.
     192            0 :     fn close_wal_store(&mut self) {
     193            0 :         if let StateSK::Loaded(sk) = self {
     194            0 :             sk.wal_store.close();
     195            0 :         }
     196            0 :     }
     197              : 
     198              :     /// Update timeline state with peer safekeeper data.
     199            0 :     pub async fn record_safekeeper_info(&mut self, sk_info: &SafekeeperTimelineInfo) -> Result<()> {
     200            0 :         // update commit_lsn if safekeeper is loaded
     201            0 :         match self {
     202            0 :             StateSK::Loaded(sk) => sk.record_safekeeper_info(sk_info).await?,
     203            0 :             StateSK::Offloaded(_) => {}
     204            0 :             StateSK::Empty => unreachable!(),
     205              :         }
     206              : 
     207              :         // update everything else, including remote_consistent_lsn and backup_lsn
     208            0 :         let mut sync_control_file = false;
     209            0 :         let state = self.state_mut();
     210            0 :         let wal_seg_size = state.server.wal_seg_size as u64;
     211            0 : 
     212            0 :         state.inmem.backup_lsn = max(Lsn(sk_info.backup_lsn), state.inmem.backup_lsn);
     213            0 :         sync_control_file |= state.backup_lsn + wal_seg_size < state.inmem.backup_lsn;
     214            0 : 
     215            0 :         state.inmem.remote_consistent_lsn = max(
     216            0 :             Lsn(sk_info.remote_consistent_lsn),
     217            0 :             state.inmem.remote_consistent_lsn,
     218            0 :         );
     219            0 :         sync_control_file |=
     220            0 :             state.remote_consistent_lsn + wal_seg_size < state.inmem.remote_consistent_lsn;
     221            0 : 
     222            0 :         state.inmem.peer_horizon_lsn =
     223            0 :             max(Lsn(sk_info.peer_horizon_lsn), state.inmem.peer_horizon_lsn);
     224            0 :         sync_control_file |= state.peer_horizon_lsn + wal_seg_size < state.inmem.peer_horizon_lsn;
     225            0 : 
     226            0 :         if sync_control_file {
     227            0 :             state.flush().await?;
     228            0 :         }
     229            0 :         Ok(())
     230            0 :     }
     231              : 
     232              :     /// Previously known as epoch_start_lsn. Needed only for reference in some APIs.
     233            0 :     pub fn term_start_lsn(&self) -> Lsn {
     234            0 :         match self {
     235            0 :             StateSK::Loaded(sk) => sk.term_start_lsn,
     236            0 :             StateSK::Offloaded(_) => Lsn(0),
     237            0 :             StateSK::Empty => unreachable!(),
     238              :         }
     239            0 :     }
     240              : 
     241              :     /// Used for metrics only.
     242            0 :     pub fn wal_storage_metrics(&self) -> WalStorageMetrics {
     243            0 :         match self {
     244            0 :             StateSK::Loaded(sk) => sk.wal_store.get_metrics(),
     245            0 :             StateSK::Offloaded(_) => WalStorageMetrics::default(),
     246            0 :             StateSK::Empty => unreachable!(),
     247              :         }
     248            0 :     }
     249              : 
     250              :     /// Returns WAL storage internal LSNs for debug dump.
     251            0 :     pub fn wal_storage_internal_state(&self) -> (Lsn, Lsn, Lsn, bool) {
     252            0 :         match self {
     253            0 :             StateSK::Loaded(sk) => sk.wal_store.internal_state(),
     254              :             StateSK::Offloaded(_) => {
     255            0 :                 let flush_lsn = self.flush_lsn();
     256            0 :                 (flush_lsn, flush_lsn, flush_lsn, false)
     257              :             }
     258            0 :             StateSK::Empty => unreachable!(),
     259              :         }
     260            0 :     }
     261              : 
     262              :     /// Access to SafeKeeper object. Panics if offloaded, should be good to use from WalResidentTimeline.
     263            0 :     pub fn safekeeper(
     264            0 :         &mut self,
     265            0 :     ) -> &mut SafeKeeper<control_file::FileStorage, wal_storage::PhysicalStorage> {
     266            0 :         match self {
     267            0 :             StateSK::Loaded(sk) => sk,
     268              :             StateSK::Offloaded(_) => {
     269            0 :                 panic!("safekeeper is offloaded, cannot be used")
     270              :             }
     271            0 :             StateSK::Empty => unreachable!(),
     272              :         }
     273            0 :     }
     274              : 
     275              :     /// Moves control file's state structure out of the enum. Used to switch states.
     276            0 :     fn take_state(self) -> TimelineState<control_file::FileStorage> {
     277            0 :         match self {
     278            0 :             StateSK::Loaded(sk) => sk.state,
     279            0 :             StateSK::Offloaded(state) => *state,
     280            0 :             StateSK::Empty => unreachable!(),
     281              :         }
     282            0 :     }
     283              : }
     284              : 
     285              : /// Shared state associated with database instance
     286              : pub struct SharedState {
     287              :     /// Safekeeper object
     288              :     pub(crate) sk: StateSK,
     289              :     /// In memory list containing state of peers sent in latest messages from them.
     290              :     pub(crate) peers_info: PeersInfo,
     291              :     // True value hinders old WAL removal; this is used by snapshotting. We
     292              :     // could make it a counter, but there is no need to.
     293              :     pub(crate) wal_removal_on_hold: bool,
     294              : }
     295              : 
     296              : impl SharedState {
     297              :     /// Creates a new SharedState.
     298            0 :     pub fn new(sk: StateSK) -> Self {
     299            0 :         Self {
     300            0 :             sk,
     301            0 :             peers_info: PeersInfo(vec![]),
     302            0 :             wal_removal_on_hold: false,
     303            0 :         }
     304            0 :     }
     305              : 
     306              :     /// Restore SharedState from control file. If file doesn't exist, bails out.
     307            0 :     pub fn restore(conf: &SafeKeeperConf, ttid: &TenantTimelineId) -> Result<Self> {
     308            0 :         let timeline_dir = get_timeline_dir(conf, ttid);
     309            0 :         let control_store = control_file::FileStorage::restore_new(&timeline_dir, conf.no_sync)?;
     310            0 :         if control_store.server.wal_seg_size == 0 {
     311            0 :             bail!(TimelineError::UninitializedWalSegSize(*ttid));
     312            0 :         }
     313              : 
     314            0 :         let sk = match control_store.eviction_state {
     315              :             EvictionState::Present => {
     316            0 :                 let wal_store = wal_storage::PhysicalStorage::new(
     317            0 :                     ttid,
     318            0 :                     &timeline_dir,
     319            0 :                     &control_store,
     320            0 :                     conf.no_sync,
     321            0 :                 )?;
     322            0 :                 StateSK::Loaded(SafeKeeper::new(
     323            0 :                     TimelineState::new(control_store),
     324            0 :                     wal_store,
     325            0 :                     conf.my_id,
     326            0 :                 )?)
     327              :             }
     328              :             EvictionState::Offloaded(_) => {
     329            0 :                 StateSK::Offloaded(Box::new(TimelineState::new(control_store)))
     330              :             }
     331              :         };
     332              : 
     333            0 :         Ok(Self::new(sk))
     334            0 :     }
     335              : 
     336            0 :     pub(crate) fn get_wal_seg_size(&self) -> usize {
     337            0 :         self.sk.state().server.wal_seg_size as usize
     338            0 :     }
     339              : 
     340            0 :     fn get_safekeeper_info(
     341            0 :         &self,
     342            0 :         ttid: &TenantTimelineId,
     343            0 :         conf: &SafeKeeperConf,
     344            0 :         standby_apply_lsn: Lsn,
     345            0 :     ) -> SafekeeperTimelineInfo {
     346            0 :         SafekeeperTimelineInfo {
     347            0 :             safekeeper_id: conf.my_id.0,
     348            0 :             tenant_timeline_id: Some(ProtoTenantTimelineId {
     349            0 :                 tenant_id: ttid.tenant_id.as_ref().to_owned(),
     350            0 :                 timeline_id: ttid.timeline_id.as_ref().to_owned(),
     351            0 :             }),
     352            0 :             term: self.sk.state().acceptor_state.term,
     353            0 :             last_log_term: self.sk.last_log_term(),
     354            0 :             flush_lsn: self.sk.flush_lsn().0,
     355            0 :             // note: this value is not flushed to control file yet and can be lost
     356            0 :             commit_lsn: self.sk.state().inmem.commit_lsn.0,
     357            0 :             remote_consistent_lsn: self.sk.state().inmem.remote_consistent_lsn.0,
     358            0 :             peer_horizon_lsn: self.sk.state().inmem.peer_horizon_lsn.0,
     359            0 :             safekeeper_connstr: conf
     360            0 :                 .advertise_pg_addr
     361            0 :                 .to_owned()
     362            0 :                 .unwrap_or(conf.listen_pg_addr.clone()),
     363            0 :             http_connstr: conf.listen_http_addr.to_owned(),
     364            0 :             backup_lsn: self.sk.state().inmem.backup_lsn.0,
     365            0 :             local_start_lsn: self.sk.state().local_start_lsn.0,
     366            0 :             availability_zone: conf.availability_zone.clone(),
     367            0 :             standby_horizon: standby_apply_lsn.0,
     368            0 :         }
     369            0 :     }
     370              : 
     371              :     /// Get our latest view of alive peers status on the timeline.
     372              :     /// We pass our own info through the broker as well, so when we don't have connection
     373              :     /// to the broker returned vec is empty.
     374            0 :     pub(crate) fn get_peers(&self, heartbeat_timeout: Duration) -> Vec<PeerInfo> {
     375            0 :         let now = Instant::now();
     376            0 :         self.peers_info
     377            0 :             .0
     378            0 :             .iter()
     379            0 :             // Regard peer as absent if we haven't heard from it within heartbeat_timeout.
     380            0 :             .filter(|p| now.duration_since(p.ts) <= heartbeat_timeout)
     381            0 :             .cloned()
     382            0 :             .collect()
     383            0 :     }
     384              : }
     385              : 
     386              : #[derive(Debug, thiserror::Error)]
     387              : pub enum TimelineError {
     388              :     #[error("Timeline {0} was cancelled and cannot be used anymore")]
     389              :     Cancelled(TenantTimelineId),
     390              :     #[error("Timeline {0} was not found in global map")]
     391              :     NotFound(TenantTimelineId),
     392              :     #[error("Timeline {0} creation is in progress")]
     393              :     CreationInProgress(TenantTimelineId),
     394              :     #[error("Timeline {0} exists on disk, but wasn't loaded on startup")]
     395              :     Invalid(TenantTimelineId),
     396              :     #[error("Timeline {0} is already exists")]
     397              :     AlreadyExists(TenantTimelineId),
     398              :     #[error("Timeline {0} is not initialized, wal_seg_size is zero")]
     399              :     UninitializedWalSegSize(TenantTimelineId),
     400              :     #[error("Timeline {0} is not initialized, pg_version is unknown")]
     401              :     UninitialinzedPgVersion(TenantTimelineId),
     402              : }
     403              : 
     404              : // Convert to HTTP API error.
     405              : impl From<TimelineError> for ApiError {
     406            0 :     fn from(te: TimelineError) -> ApiError {
     407            0 :         match te {
     408            0 :             TimelineError::NotFound(ttid) => {
     409            0 :                 ApiError::NotFound(anyhow!("timeline {} not found", ttid).into())
     410              :             }
     411            0 :             _ => ApiError::InternalServerError(anyhow!("{}", te)),
     412              :         }
     413            0 :     }
     414              : }
     415              : 
     416              : /// Timeline struct manages lifecycle (creation, deletion, restore) of a safekeeper timeline.
     417              : /// It also holds SharedState and provides mutually exclusive access to it.
     418              : pub struct Timeline {
     419              :     pub ttid: TenantTimelineId,
     420              :     pub remote_path: RemotePath,
     421              : 
     422              :     /// Used to broadcast commit_lsn updates to all background jobs.
     423              :     commit_lsn_watch_tx: watch::Sender<Lsn>,
     424              :     commit_lsn_watch_rx: watch::Receiver<Lsn>,
     425              : 
     426              :     /// Broadcasts (current term, flush_lsn) updates, walsender is interested in
     427              :     /// them when sending in recovery mode (to walproposer or peers). Note: this
     428              :     /// is just a notification, WAL reading should always done with lock held as
     429              :     /// term can change otherwise.
     430              :     term_flush_lsn_watch_tx: watch::Sender<TermLsn>,
     431              :     term_flush_lsn_watch_rx: watch::Receiver<TermLsn>,
     432              : 
     433              :     /// Broadcasts shared state updates.
     434              :     shared_state_version_tx: watch::Sender<usize>,
     435              :     shared_state_version_rx: watch::Receiver<usize>,
     436              : 
     437              :     /// Safekeeper and other state, that should remain consistent and
     438              :     /// synchronized with the disk. This is tokio mutex as we write WAL to disk
     439              :     /// while holding it, ensuring that consensus checks are in order.
     440              :     mutex: RwLock<SharedState>,
     441              :     walsenders: Arc<WalSenders>,
     442              :     walreceivers: Arc<WalReceivers>,
     443              :     timeline_dir: Utf8PathBuf,
     444              :     manager_ctl: ManagerCtl,
     445              :     conf: Arc<SafeKeeperConf>,
     446              : 
     447              :     /// Hold this gate from code that depends on the Timeline's non-shut-down state.  While holding
     448              :     /// this gate, you must respect [`Timeline::cancel`]
     449              :     pub(crate) gate: Gate,
     450              : 
     451              :     /// Delete/cancel will trigger this, background tasks should drop out as soon as it fires
     452              :     pub(crate) cancel: CancellationToken,
     453              : 
     454              :     // timeline_manager controlled state
     455              :     pub(crate) broker_active: AtomicBool,
     456              :     pub(crate) wal_backup_active: AtomicBool,
     457              :     pub(crate) last_removed_segno: AtomicU64,
     458              :     pub(crate) mgr_status: AtomicStatus,
     459              : }
     460              : 
     461              : impl Timeline {
     462              :     /// Constructs a new timeline.
     463            0 :     pub fn new(
     464            0 :         ttid: TenantTimelineId,
     465            0 :         timeline_dir: &Utf8Path,
     466            0 :         remote_path: &RemotePath,
     467            0 :         shared_state: SharedState,
     468            0 :         conf: Arc<SafeKeeperConf>,
     469            0 :     ) -> Arc<Self> {
     470            0 :         let (commit_lsn_watch_tx, commit_lsn_watch_rx) =
     471            0 :             watch::channel(shared_state.sk.state().commit_lsn);
     472            0 :         let (term_flush_lsn_watch_tx, term_flush_lsn_watch_rx) = watch::channel(TermLsn::from((
     473            0 :             shared_state.sk.last_log_term(),
     474            0 :             shared_state.sk.flush_lsn(),
     475            0 :         )));
     476            0 :         let (shared_state_version_tx, shared_state_version_rx) = watch::channel(0);
     477            0 : 
     478            0 :         let walreceivers = WalReceivers::new();
     479            0 : 
     480            0 :         Arc::new(Self {
     481            0 :             ttid,
     482            0 :             remote_path: remote_path.to_owned(),
     483            0 :             timeline_dir: timeline_dir.to_owned(),
     484            0 :             commit_lsn_watch_tx,
     485            0 :             commit_lsn_watch_rx,
     486            0 :             term_flush_lsn_watch_tx,
     487            0 :             term_flush_lsn_watch_rx,
     488            0 :             shared_state_version_tx,
     489            0 :             shared_state_version_rx,
     490            0 :             mutex: RwLock::new(shared_state),
     491            0 :             walsenders: WalSenders::new(walreceivers.clone()),
     492            0 :             walreceivers,
     493            0 :             gate: Default::default(),
     494            0 :             cancel: CancellationToken::default(),
     495            0 :             manager_ctl: ManagerCtl::new(),
     496            0 :             conf,
     497            0 :             broker_active: AtomicBool::new(false),
     498            0 :             wal_backup_active: AtomicBool::new(false),
     499            0 :             last_removed_segno: AtomicU64::new(0),
     500            0 :             mgr_status: AtomicStatus::new(),
     501            0 :         })
     502            0 :     }
     503              : 
     504              :     /// Load existing timeline from disk.
     505            0 :     pub fn load_timeline(
     506            0 :         conf: Arc<SafeKeeperConf>,
     507            0 :         ttid: TenantTimelineId,
     508            0 :     ) -> Result<Arc<Timeline>> {
     509            0 :         let _enter = info_span!("load_timeline", timeline = %ttid.timeline_id).entered();
     510              : 
     511            0 :         let shared_state = SharedState::restore(conf.as_ref(), &ttid)?;
     512            0 :         let timeline_dir = get_timeline_dir(conf.as_ref(), &ttid);
     513            0 :         let remote_path = remote_timeline_path(&ttid)?;
     514              : 
     515            0 :         Ok(Timeline::new(
     516            0 :             ttid,
     517            0 :             &timeline_dir,
     518            0 :             &remote_path,
     519            0 :             shared_state,
     520            0 :             conf,
     521            0 :         ))
     522            0 :     }
     523              : 
     524              :     /// Bootstrap new or existing timeline starting background tasks.
     525            0 :     pub fn bootstrap(
     526            0 :         self: &Arc<Timeline>,
     527            0 :         _shared_state: &mut WriteGuardSharedState<'_>,
     528            0 :         conf: &SafeKeeperConf,
     529            0 :         broker_active_set: Arc<TimelinesSet>,
     530            0 :         partial_backup_rate_limiter: RateLimiter,
     531            0 :     ) {
     532            0 :         let (tx, rx) = self.manager_ctl.bootstrap_manager();
     533              : 
     534            0 :         let Ok(gate_guard) = self.gate.enter() else {
     535              :             // Init raced with shutdown
     536            0 :             return;
     537              :         };
     538              : 
     539              :         // Start manager task which will monitor timeline state and update
     540              :         // background tasks.
     541            0 :         tokio::spawn({
     542            0 :             let this = self.clone();
     543            0 :             let conf = conf.clone();
     544            0 :             async move {
     545            0 :                 let _gate_guard = gate_guard;
     546            0 :                 timeline_manager::main_task(
     547            0 :                     ManagerTimeline { tli: this },
     548            0 :                     conf,
     549            0 :                     broker_active_set,
     550            0 :                     tx,
     551            0 :                     rx,
     552            0 :                     partial_backup_rate_limiter,
     553            0 :                 )
     554            0 :                 .await
     555            0 :             }
     556            0 :         });
     557            0 :     }
     558              : 
     559              :     /// Background timeline activities (which hold Timeline::gate) will no
     560              :     /// longer run once this function completes.
     561            0 :     pub async fn shutdown(&self) {
     562            0 :         info!("timeline {} shutting down", self.ttid);
     563            0 :         self.cancel.cancel();
     564            0 : 
     565            0 :         // Wait for any concurrent tasks to stop using this timeline, to avoid e.g. attempts
     566            0 :         // to read deleted files.
     567            0 :         self.gate.close().await;
     568            0 :     }
     569              : 
     570              :     /// Delete timeline from disk completely, by removing timeline directory.
     571              :     ///
     572              :     /// Also deletes WAL in s3. Might fail if e.g. s3 is unavailable, but
     573              :     /// deletion API endpoint is retriable.
     574              :     ///
     575              :     /// Timeline must be in shut-down state (i.e. call [`Self::shutdown`] first)
     576            0 :     pub async fn delete(
     577            0 :         &self,
     578            0 :         shared_state: &mut WriteGuardSharedState<'_>,
     579            0 :         only_local: bool,
     580            0 :     ) -> Result<bool> {
     581            0 :         // Assert that [`Self::shutdown`] was already called
     582            0 :         assert!(self.cancel.is_cancelled());
     583            0 :         assert!(self.gate.close_complete());
     584              : 
     585              :         // Close associated FDs. Nobody will be able to touch timeline data once
     586              :         // it is cancelled, so WAL storage won't be opened again.
     587            0 :         shared_state.sk.close_wal_store();
     588            0 : 
     589            0 :         if !only_local && self.conf.is_wal_backup_enabled() {
     590              :             // Note: we concurrently delete remote storage data from multiple
     591              :             // safekeepers. That's ok, s3 replies 200 if object doesn't exist and we
     592              :             // do some retries anyway.
     593            0 :             wal_backup::delete_timeline(&self.ttid).await?;
     594            0 :         }
     595            0 :         let dir_existed = delete_dir(&self.timeline_dir).await?;
     596            0 :         Ok(dir_existed)
     597            0 :     }
     598              : 
     599              :     /// Returns if timeline is cancelled.
     600            0 :     pub fn is_cancelled(&self) -> bool {
     601            0 :         self.cancel.is_cancelled()
     602            0 :     }
     603              : 
     604              :     /// Take a writing mutual exclusive lock on timeline shared_state.
     605            0 :     pub async fn write_shared_state<'a>(self: &'a Arc<Self>) -> WriteGuardSharedState<'a> {
     606            0 :         WriteGuardSharedState::new(self.clone(), self.mutex.write().await)
     607            0 :     }
     608              : 
     609            0 :     pub async fn read_shared_state(&self) -> ReadGuardSharedState {
     610            0 :         self.mutex.read().await
     611            0 :     }
     612              : 
     613              :     /// Returns commit_lsn watch channel.
     614            0 :     pub fn get_commit_lsn_watch_rx(&self) -> watch::Receiver<Lsn> {
     615            0 :         self.commit_lsn_watch_rx.clone()
     616            0 :     }
     617              : 
     618              :     /// Returns term_flush_lsn watch channel.
     619            0 :     pub fn get_term_flush_lsn_watch_rx(&self) -> watch::Receiver<TermLsn> {
     620            0 :         self.term_flush_lsn_watch_rx.clone()
     621            0 :     }
     622              : 
     623              :     /// Returns watch channel for SharedState update version.
     624            0 :     pub fn get_state_version_rx(&self) -> watch::Receiver<usize> {
     625            0 :         self.shared_state_version_rx.clone()
     626            0 :     }
     627              : 
     628              :     /// Returns wal_seg_size.
     629            0 :     pub async fn get_wal_seg_size(&self) -> usize {
     630            0 :         self.read_shared_state().await.get_wal_seg_size()
     631            0 :     }
     632              : 
     633              :     /// Returns state of the timeline.
     634            0 :     pub async fn get_state(&self) -> (TimelineMemState, TimelinePersistentState) {
     635            0 :         let state = self.read_shared_state().await;
     636            0 :         (
     637            0 :             state.sk.state().inmem.clone(),
     638            0 :             TimelinePersistentState::clone(state.sk.state()),
     639            0 :         )
     640            0 :     }
     641              : 
     642              :     /// Returns latest backup_lsn.
     643            0 :     pub async fn get_wal_backup_lsn(&self) -> Lsn {
     644            0 :         self.read_shared_state().await.sk.state().inmem.backup_lsn
     645            0 :     }
     646              : 
     647              :     /// Sets backup_lsn to the given value.
     648            0 :     pub async fn set_wal_backup_lsn(self: &Arc<Self>, backup_lsn: Lsn) -> Result<()> {
     649            0 :         if self.is_cancelled() {
     650            0 :             bail!(TimelineError::Cancelled(self.ttid));
     651            0 :         }
     652              : 
     653            0 :         let mut state = self.write_shared_state().await;
     654            0 :         state.sk.state_mut().inmem.backup_lsn = max(state.sk.state().inmem.backup_lsn, backup_lsn);
     655            0 :         // we should check whether to shut down offloader, but this will be done
     656            0 :         // soon by peer communication anyway.
     657            0 :         Ok(())
     658            0 :     }
     659              : 
     660              :     /// Get safekeeper info for broadcasting to broker and other peers.
     661            0 :     pub async fn get_safekeeper_info(&self, conf: &SafeKeeperConf) -> SafekeeperTimelineInfo {
     662            0 :         let standby_apply_lsn = self.walsenders.get_hotstandby().reply.apply_lsn;
     663            0 :         let shared_state = self.read_shared_state().await;
     664            0 :         shared_state.get_safekeeper_info(&self.ttid, conf, standby_apply_lsn)
     665            0 :     }
     666              : 
     667              :     /// Update timeline state with peer safekeeper data.
     668            0 :     pub async fn record_safekeeper_info(
     669            0 :         self: &Arc<Self>,
     670            0 :         sk_info: SafekeeperTimelineInfo,
     671            0 :     ) -> Result<()> {
     672              :         {
     673            0 :             let mut shared_state = self.write_shared_state().await;
     674            0 :             shared_state.sk.record_safekeeper_info(&sk_info).await?;
     675            0 :             let peer_info = peer_info_from_sk_info(&sk_info, Instant::now());
     676            0 :             shared_state.peers_info.upsert(&peer_info);
     677            0 :         }
     678            0 :         Ok(())
     679            0 :     }
     680              : 
     681            0 :     pub async fn get_peers(&self, conf: &SafeKeeperConf) -> Vec<PeerInfo> {
     682            0 :         let shared_state = self.read_shared_state().await;
     683            0 :         shared_state.get_peers(conf.heartbeat_timeout)
     684            0 :     }
     685              : 
     686            0 :     pub fn get_walsenders(&self) -> &Arc<WalSenders> {
     687            0 :         &self.walsenders
     688            0 :     }
     689              : 
     690            0 :     pub fn get_walreceivers(&self) -> &Arc<WalReceivers> {
     691            0 :         &self.walreceivers
     692            0 :     }
     693              : 
     694              :     /// Returns flush_lsn.
     695            0 :     pub async fn get_flush_lsn(&self) -> Lsn {
     696            0 :         self.read_shared_state().await.sk.flush_lsn()
     697            0 :     }
     698              : 
     699              :     /// Gather timeline data for metrics.
     700            0 :     pub async fn info_for_metrics(&self) -> Option<FullTimelineInfo> {
     701            0 :         if self.is_cancelled() {
     702            0 :             return None;
     703            0 :         }
     704            0 : 
     705            0 :         let (ps_feedback_count, last_ps_feedback) = self.walsenders.get_ps_feedback_stats();
     706            0 :         let state = self.read_shared_state().await;
     707            0 :         Some(FullTimelineInfo {
     708            0 :             ttid: self.ttid,
     709            0 :             ps_feedback_count,
     710            0 :             last_ps_feedback,
     711            0 :             wal_backup_active: self.wal_backup_active.load(Ordering::Relaxed),
     712            0 :             timeline_is_active: self.broker_active.load(Ordering::Relaxed),
     713            0 :             num_computes: self.walreceivers.get_num() as u32,
     714            0 :             last_removed_segno: self.last_removed_segno.load(Ordering::Relaxed),
     715            0 :             epoch_start_lsn: state.sk.term_start_lsn(),
     716            0 :             mem_state: state.sk.state().inmem.clone(),
     717            0 :             persisted_state: TimelinePersistentState::clone(state.sk.state()),
     718            0 :             flush_lsn: state.sk.flush_lsn(),
     719            0 :             wal_storage: state.sk.wal_storage_metrics(),
     720            0 :         })
     721            0 :     }
     722              : 
     723              :     /// Returns in-memory timeline state to build a full debug dump.
     724            0 :     pub async fn memory_dump(&self) -> debug_dump::Memory {
     725            0 :         let state = self.read_shared_state().await;
     726              : 
     727            0 :         let (write_lsn, write_record_lsn, flush_lsn, file_open) =
     728            0 :             state.sk.wal_storage_internal_state();
     729            0 : 
     730            0 :         debug_dump::Memory {
     731            0 :             is_cancelled: self.is_cancelled(),
     732            0 :             peers_info_len: state.peers_info.0.len(),
     733            0 :             walsenders: self.walsenders.get_all(),
     734            0 :             wal_backup_active: self.wal_backup_active.load(Ordering::Relaxed),
     735            0 :             active: self.broker_active.load(Ordering::Relaxed),
     736            0 :             num_computes: self.walreceivers.get_num() as u32,
     737            0 :             last_removed_segno: self.last_removed_segno.load(Ordering::Relaxed),
     738            0 :             epoch_start_lsn: state.sk.term_start_lsn(),
     739            0 :             mem_state: state.sk.state().inmem.clone(),
     740            0 :             mgr_status: self.mgr_status.get(),
     741            0 :             write_lsn,
     742            0 :             write_record_lsn,
     743            0 :             flush_lsn,
     744            0 :             file_open,
     745            0 :         }
     746            0 :     }
     747              : 
     748              :     /// Apply a function to the control file state and persist it.
     749            0 :     pub async fn map_control_file<T>(
     750            0 :         self: &Arc<Self>,
     751            0 :         f: impl FnOnce(&mut TimelinePersistentState) -> Result<T>,
     752            0 :     ) -> Result<T> {
     753            0 :         let mut state = self.write_shared_state().await;
     754            0 :         let mut persistent_state = state.sk.state_mut().start_change();
     755              :         // If f returns error, we abort the change and don't persist anything.
     756            0 :         let res = f(&mut persistent_state)?;
     757              :         // If persisting fails, we abort the change and return error.
     758            0 :         state
     759            0 :             .sk
     760            0 :             .state_mut()
     761            0 :             .finish_change(&persistent_state)
     762            0 :             .await?;
     763            0 :         Ok(res)
     764            0 :     }
     765              : 
     766            0 :     pub async fn term_bump(self: &Arc<Self>, to: Option<Term>) -> Result<TimelineTermBumpResponse> {
     767            0 :         let mut state = self.write_shared_state().await;
     768            0 :         state.sk.term_bump(to).await
     769            0 :     }
     770              : 
     771              :     /// Guts of [`Self::wal_residence_guard`] and [`Self::try_wal_residence_guard`]
     772            0 :     async fn do_wal_residence_guard(
     773            0 :         self: &Arc<Self>,
     774            0 :         block: bool,
     775            0 :     ) -> Result<Option<WalResidentTimeline>> {
     776            0 :         let op_label = if block {
     777            0 :             "wal_residence_guard"
     778              :         } else {
     779            0 :             "try_wal_residence_guard"
     780              :         };
     781              : 
     782            0 :         if self.is_cancelled() {
     783            0 :             bail!(TimelineError::Cancelled(self.ttid));
     784            0 :         }
     785            0 : 
     786            0 :         debug!("requesting WalResidentTimeline guard");
     787            0 :         let started_at = Instant::now();
     788            0 :         let status_before = self.mgr_status.get();
     789              : 
     790              :         // Wait 30 seconds for the guard to be acquired. It can time out if someone is
     791              :         // holding the lock (e.g. during `SafeKeeper::process_msg()`) or manager task
     792              :         // is stuck.
     793            0 :         let res = tokio::time::timeout_at(started_at + Duration::from_secs(30), async {
     794            0 :             if block {
     795            0 :                 self.manager_ctl.wal_residence_guard().await.map(Some)
     796              :             } else {
     797            0 :                 self.manager_ctl.try_wal_residence_guard().await
     798              :             }
     799            0 :         })
     800            0 :         .await;
     801              : 
     802            0 :         let guard = match res {
     803            0 :             Ok(Ok(guard)) => {
     804            0 :                 let finished_at = Instant::now();
     805            0 :                 let elapsed = finished_at - started_at;
     806            0 :                 MISC_OPERATION_SECONDS
     807            0 :                     .with_label_values(&[op_label])
     808            0 :                     .observe(elapsed.as_secs_f64());
     809            0 : 
     810            0 :                 guard
     811              :             }
     812            0 :             Ok(Err(e)) => {
     813            0 :                 warn!(
     814            0 :                     "error acquiring in {op_label}, statuses {:?} => {:?}",
     815            0 :                     status_before,
     816            0 :                     self.mgr_status.get()
     817              :                 );
     818            0 :                 return Err(e);
     819              :             }
     820              :             Err(_) => {
     821            0 :                 warn!(
     822            0 :                     "timeout acquiring in {op_label} guard, statuses {:?} => {:?}",
     823            0 :                     status_before,
     824            0 :                     self.mgr_status.get()
     825              :                 );
     826            0 :                 anyhow::bail!("timeout while acquiring WalResidentTimeline guard");
     827              :             }
     828              :         };
     829              : 
     830            0 :         Ok(guard.map(|g| WalResidentTimeline::new(self.clone(), g)))
     831            0 :     }
     832              : 
     833              :     /// Get the timeline guard for reading/writing WAL files.
     834              :     /// If WAL files are not present on disk (evicted), they will be automatically
     835              :     /// downloaded from remote storage. This is done in the manager task, which is
     836              :     /// responsible for issuing all guards.
     837              :     ///
     838              :     /// NB: don't use this function from timeline_manager, it will deadlock.
     839              :     /// NB: don't use this function while holding shared_state lock.
     840            0 :     pub async fn wal_residence_guard(self: &Arc<Self>) -> Result<WalResidentTimeline> {
     841            0 :         self.do_wal_residence_guard(true)
     842            0 :             .await
     843            0 :             .map(|m| m.expect("Always get Some in block=true mode"))
     844            0 :     }
     845              : 
     846              :     /// Get the timeline guard for reading/writing WAL files if the timeline is resident,
     847              :     /// else return None
     848            0 :     pub(crate) async fn try_wal_residence_guard(
     849            0 :         self: &Arc<Self>,
     850            0 :     ) -> Result<Option<WalResidentTimeline>> {
     851            0 :         self.do_wal_residence_guard(false).await
     852            0 :     }
     853              : 
     854            0 :     pub async fn backup_partial_reset(self: &Arc<Self>) -> Result<Vec<String>> {
     855            0 :         self.manager_ctl.backup_partial_reset().await
     856            0 :     }
     857              : }
     858              : 
     859              : /// This is a guard that allows to read/write disk timeline state.
     860              : /// All tasks that are trying to read/write WAL from disk should use this guard.
     861              : pub struct WalResidentTimeline {
     862              :     pub tli: Arc<Timeline>,
     863              :     _guard: ResidenceGuard,
     864              : }
     865              : 
     866              : impl WalResidentTimeline {
     867            0 :     pub fn new(tli: Arc<Timeline>, _guard: ResidenceGuard) -> Self {
     868            0 :         WalResidentTimeline { tli, _guard }
     869            0 :     }
     870              : }
     871              : 
     872              : impl Deref for WalResidentTimeline {
     873              :     type Target = Arc<Timeline>;
     874              : 
     875            0 :     fn deref(&self) -> &Self::Target {
     876            0 :         &self.tli
     877            0 :     }
     878              : }
     879              : 
     880              : impl WalResidentTimeline {
     881              :     /// Returns true if walsender should stop sending WAL to pageserver. We
     882              :     /// terminate it if remote_consistent_lsn reached commit_lsn and there is no
     883              :     /// computes. While there might be nothing to stream already, we learn about
     884              :     /// remote_consistent_lsn update through replication feedback, and we want
     885              :     /// to stop pushing to the broker if pageserver is fully caughtup.
     886            0 :     pub async fn should_walsender_stop(&self, reported_remote_consistent_lsn: Lsn) -> bool {
     887            0 :         if self.is_cancelled() {
     888            0 :             return true;
     889            0 :         }
     890            0 :         let shared_state = self.read_shared_state().await;
     891            0 :         if self.walreceivers.get_num() == 0 {
     892            0 :             return shared_state.sk.state().inmem.commit_lsn == Lsn(0) || // no data at all yet
     893            0 :             reported_remote_consistent_lsn >= shared_state.sk.state().inmem.commit_lsn;
     894            0 :         }
     895            0 :         false
     896            0 :     }
     897              : 
     898              :     /// Ensure that current term is t, erroring otherwise, and lock the state.
     899            0 :     pub async fn acquire_term(&self, t: Term) -> Result<ReadGuardSharedState> {
     900            0 :         let ss = self.read_shared_state().await;
     901            0 :         if ss.sk.state().acceptor_state.term != t {
     902            0 :             bail!(
     903            0 :                 "failed to acquire term {}, current term {}",
     904            0 :                 t,
     905            0 :                 ss.sk.state().acceptor_state.term
     906            0 :             );
     907            0 :         }
     908            0 :         Ok(ss)
     909            0 :     }
     910              : 
     911              :     /// Pass arrived message to the safekeeper.
     912            0 :     pub async fn process_msg(
     913            0 :         &self,
     914            0 :         msg: &ProposerAcceptorMessage,
     915            0 :     ) -> Result<Option<AcceptorProposerMessage>> {
     916            0 :         if self.is_cancelled() {
     917            0 :             bail!(TimelineError::Cancelled(self.ttid));
     918            0 :         }
     919              : 
     920              :         let mut rmsg: Option<AcceptorProposerMessage>;
     921              :         {
     922            0 :             let mut shared_state = self.write_shared_state().await;
     923            0 :             rmsg = shared_state.sk.safekeeper().process_msg(msg).await?;
     924              : 
     925              :             // if this is AppendResponse, fill in proper hot standby feedback.
     926            0 :             if let Some(AcceptorProposerMessage::AppendResponse(ref mut resp)) = rmsg {
     927            0 :                 resp.hs_feedback = self.walsenders.get_hotstandby().hs_feedback;
     928            0 :             }
     929              :         }
     930            0 :         Ok(rmsg)
     931            0 :     }
     932              : 
     933            0 :     pub async fn get_walreader(&self, start_lsn: Lsn) -> Result<WalReader> {
     934            0 :         let (_, persisted_state) = self.get_state().await;
     935            0 :         let enable_remote_read = self.conf.is_wal_backup_enabled();
     936            0 : 
     937            0 :         WalReader::new(
     938            0 :             &self.ttid,
     939            0 :             self.timeline_dir.clone(),
     940            0 :             &persisted_state,
     941            0 :             start_lsn,
     942            0 :             enable_remote_read,
     943            0 :         )
     944            0 :     }
     945              : 
     946            0 :     pub fn get_timeline_dir(&self) -> Utf8PathBuf {
     947            0 :         self.timeline_dir.clone()
     948            0 :     }
     949              : 
     950              :     /// Update in memory remote consistent lsn.
     951            0 :     pub async fn update_remote_consistent_lsn(&self, candidate: Lsn) {
     952            0 :         let mut shared_state = self.write_shared_state().await;
     953            0 :         shared_state.sk.state_mut().inmem.remote_consistent_lsn = max(
     954            0 :             shared_state.sk.state().inmem.remote_consistent_lsn,
     955            0 :             candidate,
     956            0 :         );
     957            0 :     }
     958              : }
     959              : 
     960              : /// This struct contains methods that are used by timeline manager task.
     961              : pub(crate) struct ManagerTimeline {
     962              :     pub(crate) tli: Arc<Timeline>,
     963              : }
     964              : 
     965              : impl Deref for ManagerTimeline {
     966              :     type Target = Arc<Timeline>;
     967              : 
     968            0 :     fn deref(&self) -> &Self::Target {
     969            0 :         &self.tli
     970            0 :     }
     971              : }
     972              : 
     973              : impl ManagerTimeline {
     974            0 :     pub(crate) fn timeline_dir(&self) -> &Utf8PathBuf {
     975            0 :         &self.tli.timeline_dir
     976            0 :     }
     977              : 
     978              :     /// Manager requests this state on startup.
     979            0 :     pub(crate) async fn bootstrap_mgr(&self) -> (bool, Option<PartialRemoteSegment>) {
     980            0 :         let shared_state = self.read_shared_state().await;
     981            0 :         let is_offloaded = matches!(
     982            0 :             shared_state.sk.state().eviction_state,
     983              :             EvictionState::Offloaded(_)
     984              :         );
     985            0 :         let partial_backup_uploaded = shared_state.sk.state().partial_backup.uploaded_segment();
     986            0 : 
     987            0 :         (is_offloaded, partial_backup_uploaded)
     988            0 :     }
     989              : 
     990              :     /// Try to switch state Present->Offloaded.
     991            0 :     pub(crate) async fn switch_to_offloaded(
     992            0 :         &self,
     993            0 :         partial: &PartialRemoteSegment,
     994            0 :     ) -> anyhow::Result<()> {
     995            0 :         let mut shared = self.write_shared_state().await;
     996              : 
     997              :         // updating control file
     998            0 :         let mut pstate = shared.sk.state_mut().start_change();
     999              : 
    1000            0 :         if !matches!(pstate.eviction_state, EvictionState::Present) {
    1001            0 :             bail!(
    1002            0 :                 "cannot switch to offloaded state, current state is {:?}",
    1003            0 :                 pstate.eviction_state
    1004            0 :             );
    1005            0 :         }
    1006            0 : 
    1007            0 :         if partial.flush_lsn != shared.sk.flush_lsn() {
    1008            0 :             bail!(
    1009            0 :                 "flush_lsn mismatch in partial backup, expected {}, got {}",
    1010            0 :                 shared.sk.flush_lsn(),
    1011            0 :                 partial.flush_lsn
    1012            0 :             );
    1013            0 :         }
    1014            0 : 
    1015            0 :         if partial.commit_lsn != pstate.commit_lsn {
    1016            0 :             bail!(
    1017            0 :                 "commit_lsn mismatch in partial backup, expected {}, got {}",
    1018            0 :                 pstate.commit_lsn,
    1019            0 :                 partial.commit_lsn
    1020            0 :             );
    1021            0 :         }
    1022            0 : 
    1023            0 :         if partial.term != shared.sk.last_log_term() {
    1024            0 :             bail!(
    1025            0 :                 "term mismatch in partial backup, expected {}, got {}",
    1026            0 :                 shared.sk.last_log_term(),
    1027            0 :                 partial.term
    1028            0 :             );
    1029            0 :         }
    1030            0 : 
    1031            0 :         pstate.eviction_state = EvictionState::Offloaded(shared.sk.flush_lsn());
    1032            0 :         shared.sk.state_mut().finish_change(&pstate).await?;
    1033              :         // control file is now switched to Offloaded state
    1034              : 
    1035              :         // now we can switch shared.sk to Offloaded, shouldn't fail
    1036            0 :         let prev_sk = std::mem::replace(&mut shared.sk, StateSK::Empty);
    1037            0 :         let cfile_state = prev_sk.take_state();
    1038            0 :         shared.sk = StateSK::Offloaded(Box::new(cfile_state));
    1039            0 : 
    1040            0 :         Ok(())
    1041            0 :     }
    1042              : 
    1043              :     /// Try to switch state Offloaded->Present.
    1044            0 :     pub(crate) async fn switch_to_present(&self) -> anyhow::Result<()> {
    1045            0 :         let mut shared = self.write_shared_state().await;
    1046              : 
    1047              :         // trying to restore WAL storage
    1048            0 :         let wal_store = wal_storage::PhysicalStorage::new(
    1049            0 :             &self.ttid,
    1050            0 :             &self.timeline_dir,
    1051            0 :             shared.sk.state(),
    1052            0 :             self.conf.no_sync,
    1053            0 :         )?;
    1054              : 
    1055              :         // updating control file
    1056            0 :         let mut pstate = shared.sk.state_mut().start_change();
    1057              : 
    1058            0 :         if !matches!(pstate.eviction_state, EvictionState::Offloaded(_)) {
    1059            0 :             bail!(
    1060            0 :                 "cannot switch to present state, current state is {:?}",
    1061            0 :                 pstate.eviction_state
    1062            0 :             );
    1063            0 :         }
    1064            0 : 
    1065            0 :         if wal_store.flush_lsn() != shared.sk.flush_lsn() {
    1066            0 :             bail!(
    1067            0 :                 "flush_lsn mismatch in restored WAL, expected {}, got {}",
    1068            0 :                 shared.sk.flush_lsn(),
    1069            0 :                 wal_store.flush_lsn()
    1070            0 :             );
    1071            0 :         }
    1072            0 : 
    1073            0 :         pstate.eviction_state = EvictionState::Present;
    1074            0 :         shared.sk.state_mut().finish_change(&pstate).await?;
    1075              : 
    1076              :         // now we can switch shared.sk to Present, shouldn't fail
    1077            0 :         let prev_sk = std::mem::replace(&mut shared.sk, StateSK::Empty);
    1078            0 :         let cfile_state = prev_sk.take_state();
    1079            0 :         shared.sk = StateSK::Loaded(SafeKeeper::new(cfile_state, wal_store, self.conf.my_id)?);
    1080              : 
    1081            0 :         Ok(())
    1082            0 :     }
    1083              : 
    1084              :     /// Update current manager state, useful for debugging manager deadlocks.
    1085            0 :     pub(crate) fn set_status(&self, status: timeline_manager::Status) {
    1086            0 :         self.mgr_status.store(status, Ordering::Relaxed);
    1087            0 :     }
    1088              : }
    1089              : 
    1090              : /// Deletes directory and it's contents. Returns false if directory does not exist.
    1091            0 : async fn delete_dir(path: &Utf8PathBuf) -> Result<bool> {
    1092            0 :     match fs::remove_dir_all(path).await {
    1093            0 :         Ok(_) => Ok(true),
    1094            0 :         Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
    1095            0 :         Err(e) => Err(e.into()),
    1096              :     }
    1097            0 : }
    1098              : 
    1099              : /// Get a path to the tenant directory. If you just need to get a timeline directory,
    1100              : /// use WalResidentTimeline::get_timeline_dir instead.
    1101            0 : pub fn get_tenant_dir(conf: &SafeKeeperConf, tenant_id: &TenantId) -> Utf8PathBuf {
    1102            0 :     conf.workdir.join(tenant_id.to_string())
    1103            0 : }
    1104              : 
    1105              : /// Get a path to the timeline directory. If you need to read WAL files from disk,
    1106              : /// use WalResidentTimeline::get_timeline_dir instead. This function does not check
    1107              : /// timeline eviction status and WAL files might not be present on disk.
    1108            0 : pub fn get_timeline_dir(conf: &SafeKeeperConf, ttid: &TenantTimelineId) -> Utf8PathBuf {
    1109            0 :     get_tenant_dir(conf, &ttid.tenant_id).join(ttid.timeline_id.to_string())
    1110            0 : }
        

Generated by: LCOV version 2.1-beta