LCOV - code coverage report
Current view: top level - safekeeper/src - timelines_global_map.rs (source / functions) Coverage Total Hit
Test: 1e20c4f2b28aa592527961bb32170ebbd2c9172f.info Lines: 0.0 % 371 0
Test Date: 2025-07-16 12:29:03 Functions: 0.0 % 42 0

            Line data    Source code
       1              : //! This module contains global `(tenant_id, timeline_id)` -> `Arc<Timeline>` mapping.
       2              : //! All timelines should always be present in this map, this is done by loading them
       3              : //! all from the disk on startup and keeping them in memory.
       4              : 
       5              : use std::collections::HashMap;
       6              : use std::str::FromStr;
       7              : use std::sync::{Arc, Mutex};
       8              : use std::time::{Duration, Instant};
       9              : 
      10              : use anyhow::{Context, Result, bail};
      11              : use camino::Utf8PathBuf;
      12              : use camino_tempfile::Utf8TempDir;
      13              : use safekeeper_api::membership::Configuration;
      14              : use safekeeper_api::models::{SafekeeperUtilization, TimelineDeleteResult};
      15              : use safekeeper_api::{ServerInfo, membership};
      16              : use tokio::fs;
      17              : use tracing::*;
      18              : use utils::crashsafe::{durable_rename, fsync_async_opt};
      19              : use utils::id::{TenantId, TenantTimelineId, TimelineId};
      20              : use utils::lsn::Lsn;
      21              : 
      22              : use crate::defaults::DEFAULT_EVICTION_CONCURRENCY;
      23              : use crate::http::routes::DeleteOrExcludeError;
      24              : use crate::rate_limit::RateLimiter;
      25              : use crate::state::TimelinePersistentState;
      26              : use crate::timeline::{Timeline, TimelineError, delete_dir, get_tenant_dir, get_timeline_dir};
      27              : use crate::timelines_set::TimelinesSet;
      28              : use crate::wal_backup::WalBackup;
      29              : use crate::wal_storage::Storage;
      30              : use crate::{SafeKeeperConf, control_file, wal_storage};
      31              : 
      32              : // Timeline entry in the global map: either a ready timeline, or mark that it is
      33              : // being created.
      34              : #[derive(Clone)]
      35              : enum GlobalMapTimeline {
      36              :     CreationInProgress,
      37              :     Timeline(Arc<Timeline>),
      38              : }
      39              : 
      40              : struct GlobalTimelinesState {
      41              :     timelines: HashMap<TenantTimelineId, GlobalMapTimeline>,
      42              : 
      43              :     // A tombstone indicates this timeline used to exist has been deleted.  These are used to prevent
      44              :     // on-demand timeline creation from recreating deleted timelines.  This is only soft-enforced, as
      45              :     // this map is dropped on restart.
      46              :     tombstones: HashMap<TenantTimelineId, Instant>,
      47              :     tenant_tombstones: HashMap<TenantId, Instant>,
      48              : 
      49              :     conf: Arc<SafeKeeperConf>,
      50              :     broker_active_set: Arc<TimelinesSet>,
      51              :     global_rate_limiter: RateLimiter,
      52              :     wal_backup: Arc<WalBackup>,
      53              : }
      54              : 
      55              : impl GlobalTimelinesState {
      56              :     /// Get dependencies for a timeline constructor.
      57            0 :     fn get_dependencies(
      58            0 :         &self,
      59            0 :     ) -> (
      60            0 :         Arc<SafeKeeperConf>,
      61            0 :         Arc<TimelinesSet>,
      62            0 :         RateLimiter,
      63            0 :         Arc<WalBackup>,
      64            0 :     ) {
      65            0 :         (
      66            0 :             self.conf.clone(),
      67            0 :             self.broker_active_set.clone(),
      68            0 :             self.global_rate_limiter.clone(),
      69            0 :             self.wal_backup.clone(),
      70            0 :         )
      71            0 :     }
      72              : 
      73              :     /// Get timeline from the map. Returns error if timeline doesn't exist or
      74              :     /// creation is in progress.
      75            0 :     fn get(&self, ttid: &TenantTimelineId) -> Result<Arc<Timeline>, TimelineError> {
      76            0 :         match self.timelines.get(ttid).cloned() {
      77            0 :             Some(GlobalMapTimeline::Timeline(tli)) => Ok(tli),
      78              :             Some(GlobalMapTimeline::CreationInProgress) => {
      79            0 :                 Err(TimelineError::CreationInProgress(*ttid))
      80              :             }
      81              :             None => {
      82            0 :                 if self.has_tombstone(ttid) {
      83            0 :                     Err(TimelineError::Deleted(*ttid))
      84              :                 } else {
      85            0 :                     Err(TimelineError::NotFound(*ttid))
      86              :                 }
      87              :             }
      88              :         }
      89            0 :     }
      90              : 
      91            0 :     fn has_tombstone(&self, ttid: &TenantTimelineId) -> bool {
      92            0 :         self.tombstones.contains_key(ttid) || self.tenant_tombstones.contains_key(&ttid.tenant_id)
      93            0 :     }
      94              : 
      95              :     /// Removes all blocking tombstones for the given timeline ID.
      96              :     /// Returns `true` if there have been actual changes.
      97            0 :     fn remove_tombstone(&mut self, ttid: &TenantTimelineId) -> bool {
      98            0 :         self.tombstones.remove(ttid).is_some()
      99            0 :             || self.tenant_tombstones.remove(&ttid.tenant_id).is_some()
     100            0 :     }
     101              : 
     102            0 :     fn delete(&mut self, ttid: TenantTimelineId) {
     103            0 :         self.timelines.remove(&ttid);
     104            0 :         self.tombstones.insert(ttid, Instant::now());
     105            0 :     }
     106              : 
     107            0 :     fn add_tenant_tombstone(&mut self, tenant_id: TenantId) {
     108            0 :         self.tenant_tombstones.insert(tenant_id, Instant::now());
     109            0 :     }
     110              : }
     111              : 
     112              : /// A struct used to manage access to the global timelines map.
     113              : pub struct GlobalTimelines {
     114              :     state: Mutex<GlobalTimelinesState>,
     115              : }
     116              : 
     117              : impl GlobalTimelines {
     118              :     /// Create a new instance of the global timelines map.
     119            0 :     pub fn new(conf: Arc<SafeKeeperConf>, wal_backup: Arc<WalBackup>) -> Self {
     120            0 :         Self {
     121            0 :             state: Mutex::new(GlobalTimelinesState {
     122            0 :                 timelines: HashMap::new(),
     123            0 :                 tombstones: HashMap::new(),
     124            0 :                 tenant_tombstones: HashMap::new(),
     125            0 :                 conf,
     126            0 :                 broker_active_set: Arc::new(TimelinesSet::default()),
     127            0 :                 global_rate_limiter: RateLimiter::new(1, 1),
     128            0 :                 wal_backup,
     129            0 :             }),
     130            0 :         }
     131            0 :     }
     132              : 
     133              :     /// Inject dependencies needed for the timeline constructors and load all timelines to memory.
     134            0 :     pub async fn init(&self) -> Result<()> {
     135              :         // clippy isn't smart enough to understand that drop(state) releases the
     136              :         // lock, so use explicit block
     137            0 :         let tenants_dir = {
     138            0 :             let mut state = self.state.lock().unwrap();
     139            0 :             state.global_rate_limiter = RateLimiter::new(
     140            0 :                 state.conf.partial_backup_concurrency,
     141              :                 DEFAULT_EVICTION_CONCURRENCY,
     142              :             );
     143              : 
     144              :             // Iterate through all directories and load tenants for all directories
     145              :             // named as a valid tenant_id.
     146            0 :             state.conf.workdir.clone()
     147              :         };
     148            0 :         let mut tenant_count = 0;
     149            0 :         for tenants_dir_entry in std::fs::read_dir(&tenants_dir)
     150            0 :             .with_context(|| format!("failed to list tenants dir {tenants_dir}"))?
     151              :         {
     152            0 :             match &tenants_dir_entry {
     153            0 :                 Ok(tenants_dir_entry) => {
     154            0 :                     if let Ok(tenant_id) =
     155            0 :                         TenantId::from_str(tenants_dir_entry.file_name().to_str().unwrap_or(""))
     156              :                     {
     157            0 :                         tenant_count += 1;
     158            0 :                         self.load_tenant_timelines(tenant_id).await?;
     159            0 :                     }
     160              :                 }
     161            0 :                 Err(e) => error!(
     162            0 :                     "failed to list tenants dir entry {:?} in directory {}, reason: {:?}",
     163              :                     tenants_dir_entry, tenants_dir, e
     164              :                 ),
     165              :             }
     166              :         }
     167              : 
     168            0 :         info!(
     169            0 :             "found {} tenants directories, successfully loaded {} timelines",
     170              :             tenant_count,
     171            0 :             self.state.lock().unwrap().timelines.len()
     172              :         );
     173            0 :         Ok(())
     174            0 :     }
     175              : 
     176              :     /// Loads all timelines for the given tenant to memory. Returns fs::read_dir
     177              :     /// errors if any.
     178              :     ///
     179              :     /// It is async, but self.state lock is sync and there is no important
     180              :     /// reason to make it async (it is always held for a short while), so we
     181              :     /// just lock and unlock it for each timeline -- this function is called
     182              :     /// during init when nothing else is running, so this is fine.
     183            0 :     async fn load_tenant_timelines(&self, tenant_id: TenantId) -> Result<()> {
     184            0 :         let (conf, broker_active_set, partial_backup_rate_limiter, wal_backup) = {
     185            0 :             let state = self.state.lock().unwrap();
     186            0 :             state.get_dependencies()
     187            0 :         };
     188              : 
     189            0 :         let timelines_dir = get_tenant_dir(&conf, &tenant_id);
     190            0 :         for timelines_dir_entry in std::fs::read_dir(&timelines_dir)
     191            0 :             .with_context(|| format!("failed to list timelines dir {timelines_dir}"))?
     192              :         {
     193            0 :             match &timelines_dir_entry {
     194            0 :                 Ok(timeline_dir_entry) => {
     195            0 :                     if let Ok(timeline_id) =
     196            0 :                         TimelineId::from_str(timeline_dir_entry.file_name().to_str().unwrap_or(""))
     197              :                     {
     198            0 :                         let ttid = TenantTimelineId::new(tenant_id, timeline_id);
     199            0 :                         match Timeline::load_timeline(conf.clone(), ttid, wal_backup.clone()) {
     200            0 :                             Ok(tli) => {
     201            0 :                                 let mut shared_state = tli.write_shared_state().await;
     202            0 :                                 self.state
     203            0 :                                     .lock()
     204            0 :                                     .unwrap()
     205            0 :                                     .timelines
     206            0 :                                     .insert(ttid, GlobalMapTimeline::Timeline(tli.clone()));
     207            0 :                                 tli.bootstrap(
     208            0 :                                     &mut shared_state,
     209            0 :                                     &conf,
     210            0 :                                     broker_active_set.clone(),
     211            0 :                                     partial_backup_rate_limiter.clone(),
     212            0 :                                     wal_backup.clone(),
     213              :                                 );
     214              :                             }
     215              :                             // If we can't load a timeline, it's most likely because of a corrupted
     216              :                             // directory. We will log an error and won't allow to delete/recreate
     217              :                             // this timeline. The only way to fix this timeline is to repair manually
     218              :                             // and restart the safekeeper.
     219            0 :                             Err(e) => error!(
     220            0 :                                 "failed to load timeline {} for tenant {}, reason: {:?}",
     221              :                                 timeline_id, tenant_id, e
     222              :                             ),
     223              :                         }
     224            0 :                     }
     225              :                 }
     226            0 :                 Err(e) => error!(
     227            0 :                     "failed to list timelines dir entry {:?} in directory {}, reason: {:?}",
     228              :                     timelines_dir_entry, timelines_dir, e
     229              :                 ),
     230              :             }
     231              :         }
     232              : 
     233            0 :         Ok(())
     234            0 :     }
     235              : 
     236              :     /// Get the number of timelines in the map.
     237            0 :     pub fn timelines_count(&self) -> usize {
     238            0 :         self.state.lock().unwrap().timelines.len()
     239            0 :     }
     240              : 
     241              :     /// Get the global safekeeper config.
     242            0 :     pub fn get_global_config(&self) -> Arc<SafeKeeperConf> {
     243            0 :         self.state.lock().unwrap().conf.clone()
     244            0 :     }
     245              : 
     246            0 :     pub fn get_global_broker_active_set(&self) -> Arc<TimelinesSet> {
     247            0 :         self.state.lock().unwrap().broker_active_set.clone()
     248            0 :     }
     249              : 
     250            0 :     pub fn get_wal_backup(&self) -> Arc<WalBackup> {
     251            0 :         self.state.lock().unwrap().wal_backup.clone()
     252            0 :     }
     253              : 
     254              :     /// Create a new timeline with the given id. If the timeline already exists, returns
     255              :     /// an existing timeline.
     256            0 :     pub(crate) async fn create(
     257            0 :         &self,
     258            0 :         ttid: TenantTimelineId,
     259            0 :         mconf: Configuration,
     260            0 :         server_info: ServerInfo,
     261            0 :         start_lsn: Lsn,
     262            0 :         commit_lsn: Lsn,
     263            0 :     ) -> Result<Arc<Timeline>> {
     264            0 :         let (conf, _, _, _) = {
     265            0 :             let state = self.state.lock().unwrap();
     266            0 :             if let Ok(timeline) = state.get(&ttid) {
     267              :                 // Timeline already exists, return it.
     268            0 :                 return Ok(timeline);
     269            0 :             }
     270              : 
     271            0 :             if state.has_tombstone(&ttid) {
     272            0 :                 anyhow::bail!("Timeline {ttid} is deleted, refusing to recreate");
     273            0 :             }
     274              : 
     275            0 :             state.get_dependencies()
     276              :         };
     277              : 
     278            0 :         info!("creating new timeline {}", ttid);
     279              : 
     280              :         // Do on disk initialization in tmp dir.
     281            0 :         let (_tmp_dir, tmp_dir_path) = create_temp_timeline_dir(&conf, ttid).await?;
     282              : 
     283              :         // TODO: currently we create only cfile. It would be reasonable to
     284              :         // immediately initialize first WAL segment as well.
     285            0 :         let state = TimelinePersistentState::new(&ttid, mconf, server_info, start_lsn, commit_lsn)?;
     286            0 :         control_file::FileStorage::create_new(&tmp_dir_path, state, conf.no_sync).await?;
     287            0 :         let timeline = self.load_temp_timeline(ttid, &tmp_dir_path, true).await?;
     288            0 :         Ok(timeline)
     289            0 :     }
     290              : 
     291              :     /// Move timeline from a temp directory to the main storage, and load it to
     292              :     /// the global map. Creating timeline in this way ensures atomicity: rename
     293              :     /// is atomic, so either move of the whole datadir succeeds or it doesn't,
     294              :     /// but corrupted data dir shouldn't be possible.
     295              :     ///
     296              :     /// We'd like to avoid holding map lock while doing IO, so it's a 3 step
     297              :     /// process:
     298              :     /// 1) check the global map that timeline doesn't exist and mark that we're
     299              :     ///    creating it;
     300              :     /// 2) move the directory and load the timeline
     301              :     /// 3) take lock again and insert the timeline into the global map.
     302            0 :     pub async fn load_temp_timeline(
     303            0 :         &self,
     304            0 :         ttid: TenantTimelineId,
     305            0 :         tmp_path: &Utf8PathBuf,
     306            0 :         check_tombstone: bool,
     307            0 :     ) -> Result<Arc<Timeline>> {
     308              :         // Check for existence and mark that we're creating it.
     309            0 :         let (conf, broker_active_set, partial_backup_rate_limiter, wal_backup) = {
     310            0 :             let mut state = self.state.lock().unwrap();
     311            0 :             match state.timelines.get(&ttid) {
     312              :                 Some(GlobalMapTimeline::CreationInProgress) => {
     313            0 :                     bail!(TimelineError::CreationInProgress(ttid));
     314              :                 }
     315              :                 Some(GlobalMapTimeline::Timeline(_)) => {
     316            0 :                     bail!(TimelineError::AlreadyExists(ttid));
     317              :                 }
     318            0 :                 _ => {}
     319              :             }
     320            0 :             if check_tombstone {
     321            0 :                 if state.has_tombstone(&ttid) {
     322            0 :                     anyhow::bail!("timeline {ttid} is deleted, refusing to recreate");
     323            0 :                 }
     324              :             } else {
     325              :                 // We may be have been asked to load a timeline that was previously deleted (e.g. from `pull_timeline.rs`).  We trust
     326              :                 // that the human doing this manual intervention knows what they are doing, and remove its tombstone.
     327              :                 // It's also possible that we enter this when the tenant has been deleted, even if the timeline itself has never existed.
     328            0 :                 if state.remove_tombstone(&ttid) {
     329            0 :                     warn!("un-deleted timeline {ttid}");
     330            0 :                 }
     331              :             }
     332            0 :             state
     333            0 :                 .timelines
     334            0 :                 .insert(ttid, GlobalMapTimeline::CreationInProgress);
     335            0 :             state.get_dependencies()
     336              :         };
     337              : 
     338              :         // Do the actual move and reflect the result in the map.
     339            0 :         match GlobalTimelines::install_temp_timeline(
     340            0 :             ttid,
     341            0 :             tmp_path,
     342            0 :             conf.clone(),
     343            0 :             wal_backup.clone(),
     344              :         )
     345            0 :         .await
     346              :         {
     347            0 :             Ok(timeline) => {
     348            0 :                 let mut timeline_shared_state = timeline.write_shared_state().await;
     349            0 :                 let mut state = self.state.lock().unwrap();
     350            0 :                 assert!(matches!(
     351            0 :                     state.timelines.get(&ttid),
     352              :                     Some(GlobalMapTimeline::CreationInProgress)
     353              :                 ));
     354              : 
     355            0 :                 state
     356            0 :                     .timelines
     357            0 :                     .insert(ttid, GlobalMapTimeline::Timeline(timeline.clone()));
     358            0 :                 drop(state);
     359            0 :                 timeline.bootstrap(
     360            0 :                     &mut timeline_shared_state,
     361            0 :                     &conf,
     362            0 :                     broker_active_set,
     363            0 :                     partial_backup_rate_limiter,
     364            0 :                     wal_backup,
     365              :                 );
     366            0 :                 drop(timeline_shared_state);
     367            0 :                 Ok(timeline)
     368              :             }
     369            0 :             Err(e) => {
     370              :                 // Init failed, remove the marker from the map
     371            0 :                 let mut state = self.state.lock().unwrap();
     372            0 :                 assert!(matches!(
     373            0 :                     state.timelines.get(&ttid),
     374              :                     Some(GlobalMapTimeline::CreationInProgress)
     375              :                 ));
     376            0 :                 state.timelines.remove(&ttid);
     377            0 :                 Err(e)
     378              :             }
     379              :         }
     380            0 :     }
     381              : 
     382              :     /// Main part of load_temp_timeline: do the move and load.
     383            0 :     async fn install_temp_timeline(
     384            0 :         ttid: TenantTimelineId,
     385            0 :         tmp_path: &Utf8PathBuf,
     386            0 :         conf: Arc<SafeKeeperConf>,
     387            0 :         wal_backup: Arc<WalBackup>,
     388            0 :     ) -> Result<Arc<Timeline>> {
     389            0 :         let tenant_path = get_tenant_dir(conf.as_ref(), &ttid.tenant_id);
     390            0 :         let timeline_path = get_timeline_dir(conf.as_ref(), &ttid);
     391              : 
     392              :         // We must have already checked that timeline doesn't exist in the map,
     393              :         // but there might be existing datadir: if timeline is corrupted it is
     394              :         // not loaded. We don't want to overwrite such a dir, so check for its
     395              :         // existence.
     396            0 :         match fs::metadata(&timeline_path).await {
     397              :             Ok(_) => {
     398              :                 // Timeline directory exists on disk, we should leave state unchanged
     399              :                 // and return error.
     400            0 :                 bail!(TimelineError::Invalid(ttid));
     401              :             }
     402            0 :             Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
     403            0 :             Err(e) => {
     404            0 :                 return Err(e.into());
     405              :             }
     406              :         }
     407              : 
     408            0 :         info!(
     409            0 :             "moving timeline {} from {} to {}",
     410              :             ttid, tmp_path, timeline_path
     411              :         );
     412              : 
     413              :         // Now it is safe to move the timeline directory to the correct
     414              :         // location. First, create tenant directory. Ignore error if it already
     415              :         // exists.
     416            0 :         if let Err(e) = tokio::fs::create_dir(&tenant_path).await {
     417            0 :             if e.kind() != std::io::ErrorKind::AlreadyExists {
     418            0 :                 return Err(e.into());
     419            0 :             }
     420            0 :         }
     421              :         // fsync it
     422            0 :         fsync_async_opt(&tenant_path, !conf.no_sync).await?;
     423              :         // and its creation
     424            0 :         fsync_async_opt(&conf.workdir, !conf.no_sync).await?;
     425              : 
     426              :         // Do the move.
     427            0 :         durable_rename(tmp_path, &timeline_path, !conf.no_sync).await?;
     428              : 
     429            0 :         Timeline::load_timeline(conf, ttid, wal_backup)
     430            0 :     }
     431              : 
     432              :     /// Get a timeline from the global map. If it's not present, it doesn't exist on disk,
     433              :     /// or was corrupted and couldn't be loaded on startup. Returned timeline is always valid,
     434              :     /// i.e. loaded in memory and not cancelled.
     435            0 :     pub(crate) fn get(&self, ttid: TenantTimelineId) -> Result<Arc<Timeline>, TimelineError> {
     436            0 :         let tli_res = {
     437            0 :             let state = self.state.lock().unwrap();
     438            0 :             state.get(&ttid)
     439              :         };
     440            0 :         match tli_res {
     441            0 :             Ok(tli) => {
     442            0 :                 if tli.is_cancelled() {
     443            0 :                     return Err(TimelineError::Cancelled(ttid));
     444            0 :                 }
     445            0 :                 Ok(tli)
     446              :             }
     447            0 :             _ => tli_res,
     448              :         }
     449            0 :     }
     450              : 
     451              :     /// Returns all timelines. This is used for background timeline processes.
     452            0 :     pub fn get_all(&self) -> Vec<Arc<Timeline>> {
     453            0 :         let global_lock = self.state.lock().unwrap();
     454            0 :         global_lock
     455            0 :             .timelines
     456            0 :             .values()
     457            0 :             .filter_map(|t| match t {
     458            0 :                 GlobalMapTimeline::Timeline(t) => {
     459            0 :                     if t.is_cancelled() {
     460            0 :                         None
     461              :                     } else {
     462            0 :                         Some(t.clone())
     463              :                     }
     464              :                 }
     465            0 :                 _ => None,
     466            0 :             })
     467            0 :             .collect()
     468            0 :     }
     469              : 
     470              :     /// Returns statistics about timeline counts
     471            0 :     pub fn get_timeline_counts(&self) -> SafekeeperUtilization {
     472            0 :         let global_lock = self.state.lock().unwrap();
     473            0 :         let timeline_count = global_lock
     474            0 :             .timelines
     475            0 :             .values()
     476            0 :             .filter(|t| match t {
     477            0 :                 GlobalMapTimeline::CreationInProgress => false,
     478            0 :                 GlobalMapTimeline::Timeline(t) => !t.is_cancelled(),
     479            0 :             })
     480            0 :             .count() as u64;
     481            0 :         SafekeeperUtilization { timeline_count }
     482            0 :     }
     483              : 
     484              :     /// Returns all timelines belonging to a given tenant. Used for deleting all timelines of a tenant,
     485              :     /// and that's why it can return cancelled timelines, to retry deleting them.
     486            0 :     fn get_all_for_tenant(&self, tenant_id: TenantId) -> Vec<Arc<Timeline>> {
     487            0 :         let global_lock = self.state.lock().unwrap();
     488            0 :         global_lock
     489            0 :             .timelines
     490            0 :             .values()
     491            0 :             .filter_map(|t| match t {
     492            0 :                 GlobalMapTimeline::Timeline(t) => Some(t.clone()),
     493            0 :                 _ => None,
     494            0 :             })
     495            0 :             .filter(|t| t.ttid.tenant_id == tenant_id)
     496            0 :             .collect()
     497            0 :     }
     498              : 
     499              :     /// Delete timeline, only locally on this node or globally (also cleaning
     500              :     /// remote storage WAL), depending on `action` value.
     501            0 :     pub(crate) async fn delete_or_exclude(
     502            0 :         &self,
     503            0 :         ttid: &TenantTimelineId,
     504            0 :         action: DeleteOrExclude,
     505            0 :     ) -> Result<TimelineDeleteResult, DeleteOrExcludeError> {
     506            0 :         let tli_res = {
     507            0 :             let state = self.state.lock().unwrap();
     508              : 
     509              :             // Do NOT check tenant tombstones here: those were set earlier
     510            0 :             if state.tombstones.contains_key(ttid) {
     511              :                 // Presence of a tombstone guarantees that a previous deletion has completed and there is no work to do.
     512            0 :                 info!("Timeline {ttid} was already deleted");
     513            0 :                 return Ok(TimelineDeleteResult { dir_existed: false });
     514            0 :             }
     515              : 
     516            0 :             state.get(ttid)
     517              :         };
     518              : 
     519            0 :         let result = match tli_res {
     520            0 :             Ok(timeline) => {
     521            0 :                 info!("deleting timeline {}, action={:?}", ttid, action);
     522              : 
     523              :                 // If node is getting excluded, check the generation first.
     524              :                 // Then, while holding the lock cancel the timeline; it will be
     525              :                 // unusable after this point, and if node is added back first
     526              :                 // deletion must be completed and node seeded anew.
     527              :                 //
     528              :                 // We would like to avoid holding the lock while waiting for the
     529              :                 // gate to finish as this is deadlock prone, so for actual
     530              :                 // deletion will take it second time.
     531            0 :                 if let DeleteOrExclude::Exclude(ref mconf) = action {
     532            0 :                     let shared_state = timeline.read_shared_state().await;
     533            0 :                     if shared_state.sk.state().mconf.generation > mconf.generation {
     534            0 :                         return Err(DeleteOrExcludeError::Conflict {
     535            0 :                             requested: mconf.clone(),
     536            0 :                             current: shared_state.sk.state().mconf.clone(),
     537            0 :                         });
     538            0 :                     }
     539            0 :                     timeline.cancel().await;
     540              :                 } else {
     541            0 :                     timeline.cancel().await;
     542              :                 }
     543              : 
     544            0 :                 timeline.close().await;
     545              : 
     546            0 :                 info!("timeline {ttid} shut down for deletion");
     547              : 
     548              :                 // Take a lock and finish the deletion holding this mutex.
     549            0 :                 let mut shared_state = timeline.write_shared_state().await;
     550              : 
     551            0 :                 let only_local = !matches!(action, DeleteOrExclude::Delete);
     552            0 :                 let dir_existed = timeline.delete(&mut shared_state, only_local).await?;
     553              : 
     554            0 :                 Ok(TimelineDeleteResult { dir_existed })
     555              :             }
     556              :             Err(_) => {
     557              :                 // Timeline is not memory, but it may still exist on disk in broken state.
     558            0 :                 let dir_path = get_timeline_dir(self.state.lock().unwrap().conf.as_ref(), ttid);
     559            0 :                 let dir_existed = delete_dir(&dir_path).await?;
     560              : 
     561            0 :                 Ok(TimelineDeleteResult { dir_existed })
     562              :             }
     563              :         };
     564              : 
     565              :         // Finalize deletion, by dropping Timeline objects and storing smaller tombstones.  The tombstones
     566              :         // are used to prevent still-running computes from re-creating the same timeline when they send data,
     567              :         // and to speed up repeated deletion calls by avoiding re-listing objects.
     568            0 :         self.state.lock().unwrap().delete(*ttid);
     569              : 
     570            0 :         result
     571            0 :     }
     572              : 
     573              :     /// Deactivates and deletes all timelines for the tenant. Returns map of all timelines which
     574              :     /// the tenant had, `true` if a timeline was active. There may be a race if new timelines are
     575              :     /// created simultaneously. In that case the function will return error and the caller should
     576              :     /// retry tenant deletion again later.
     577              :     ///
     578              :     /// If only_local, doesn't remove WAL segments in remote storage.
     579            0 :     pub async fn delete_all_for_tenant(
     580            0 :         &self,
     581            0 :         tenant_id: &TenantId,
     582            0 :         action: DeleteOrExclude,
     583            0 :     ) -> Result<HashMap<TenantTimelineId, TimelineDeleteResult>> {
     584            0 :         info!("deleting all timelines for tenant {}", tenant_id);
     585              : 
     586              :         // Adding a tombstone before getting the timelines to prevent new timeline additions
     587            0 :         self.state.lock().unwrap().add_tenant_tombstone(*tenant_id);
     588              : 
     589            0 :         let to_delete = self.get_all_for_tenant(*tenant_id);
     590              : 
     591            0 :         let mut err = None;
     592              : 
     593            0 :         let mut deleted = HashMap::new();
     594            0 :         for tli in &to_delete {
     595            0 :             match self.delete_or_exclude(&tli.ttid, action.clone()).await {
     596            0 :                 Ok(result) => {
     597            0 :                     deleted.insert(tli.ttid, result);
     598            0 :                 }
     599            0 :                 Err(e) => {
     600            0 :                     error!("failed to delete timeline {}: {}", tli.ttid, e);
     601              :                     // Save error to return later.
     602            0 :                     err = Some(e);
     603              :                 }
     604              :             }
     605              :         }
     606              : 
     607              :         // If there was an error, return it.
     608            0 :         if let Some(e) = err {
     609            0 :             return Err(anyhow::Error::from(e));
     610            0 :         }
     611              : 
     612              :         // There may be broken timelines on disk, so delete the whole tenant dir as well.
     613              :         // Note that we could concurrently create new timelines while we were deleting them,
     614              :         // so the directory may be not empty. In this case timelines will have bad state
     615              :         // and timeline background jobs can panic.
     616            0 :         let tenant_dir = get_tenant_dir(self.state.lock().unwrap().conf.as_ref(), tenant_id);
     617            0 :         delete_dir(&tenant_dir).await?;
     618              : 
     619            0 :         Ok(deleted)
     620            0 :     }
     621              : 
     622            0 :     pub fn housekeeping(&self, tombstone_ttl: &Duration) {
     623            0 :         let mut state = self.state.lock().unwrap();
     624              : 
     625              :         // We keep tombstones long enough to have a good chance of preventing rogue computes from re-creating deleted
     626              :         // timelines.  If a compute kept running for longer than this TTL (or across a safekeeper restart) then they
     627              :         // may recreate a deleted timeline.
     628            0 :         let now = Instant::now();
     629            0 :         state
     630            0 :             .tombstones
     631            0 :             .retain(|_, v| now.duration_since(*v) < *tombstone_ttl);
     632            0 :         state
     633            0 :             .tenant_tombstones
     634            0 :             .retain(|_, v| now.duration_since(*v) < *tombstone_ttl);
     635            0 :     }
     636              : }
     637              : 
     638              : /// Action for delete_or_exclude.
     639              : #[derive(Clone, Debug)]
     640              : pub enum DeleteOrExclude {
     641              :     /// Delete timeline globally.
     642              :     Delete,
     643              :     /// Legacy mode until we fully migrate to generations: like exclude deletes
     644              :     /// timeline only locally, but ignores generation number.
     645              :     DeleteLocal,
     646              :     /// This node is getting excluded, delete timeline locally.
     647              :     Exclude(membership::Configuration),
     648              : }
     649              : 
     650              : /// Create temp directory for a new timeline. It needs to be located on the same
     651              : /// filesystem as the rest of the timelines. It will be automatically deleted when
     652              : /// Utf8TempDir goes out of scope.
     653            0 : pub async fn create_temp_timeline_dir(
     654            0 :     conf: &SafeKeeperConf,
     655            0 :     ttid: TenantTimelineId,
     656            0 : ) -> Result<(Utf8TempDir, Utf8PathBuf)> {
     657            0 :     let temp_base = conf.workdir.join("tmp");
     658              : 
     659            0 :     tokio::fs::create_dir_all(&temp_base).await?;
     660              : 
     661            0 :     let tli_dir = camino_tempfile::Builder::new()
     662            0 :         .suffix("_temptli")
     663            0 :         .prefix(&format!("{}_{}_", ttid.tenant_id, ttid.timeline_id))
     664            0 :         .tempdir_in(temp_base)?;
     665              : 
     666            0 :     let tli_dir_path = tli_dir.path().to_path_buf();
     667              : 
     668            0 :     Ok((tli_dir, tli_dir_path))
     669            0 : }
     670              : 
     671              : /// Do basic validation of a temp timeline, before moving it to the global map.
     672            0 : pub async fn validate_temp_timeline(
     673            0 :     conf: &SafeKeeperConf,
     674            0 :     ttid: TenantTimelineId,
     675            0 :     path: &Utf8PathBuf,
     676            0 : ) -> Result<(Lsn, Lsn)> {
     677            0 :     let control_path = path.join("safekeeper.control");
     678              : 
     679            0 :     let control_store = control_file::FileStorage::load_control_file(control_path)?;
     680            0 :     if control_store.server.wal_seg_size == 0 {
     681            0 :         bail!("wal_seg_size is not set");
     682            0 :     }
     683              : 
     684            0 :     let wal_store = wal_storage::PhysicalStorage::new(&ttid, path, &control_store, conf.no_sync)?;
     685              : 
     686            0 :     let commit_lsn = control_store.commit_lsn;
     687            0 :     let flush_lsn = wal_store.flush_lsn();
     688              : 
     689            0 :     Ok((commit_lsn, flush_lsn))
     690            0 : }
        

Generated by: LCOV version 2.1-beta