LCOV - code coverage report
Current view: top level - safekeeper/src - state.rs (source / functions) Coverage Total Hit
Test: 6df3fc19ec669bcfbbf9aba41d1338898d24eaa0.info Lines: 75.2 % 141 106
Test Date: 2025-03-12 18:28:53 Functions: 57.5 % 80 46

            Line data    Source code
       1              : //! Defines per timeline data stored persistently (SafeKeeperPersistentState)
       2              : //! and its wrapper with in memory layer (SafekeeperState).
       3              : 
       4              : use std::cmp::max;
       5              : use std::ops::Deref;
       6              : use std::time::SystemTime;
       7              : 
       8              : use anyhow::{Result, bail};
       9              : use postgres_ffi::WAL_SEGMENT_SIZE;
      10              : use safekeeper_api::membership::Configuration;
      11              : use safekeeper_api::models::{TimelineMembershipSwitchResponse, TimelineTermBumpResponse};
      12              : use safekeeper_api::{INITIAL_TERM, ServerInfo, Term};
      13              : use serde::{Deserialize, Serialize};
      14              : use tracing::info;
      15              : use utils::id::{TenantId, TenantTimelineId, TimelineId};
      16              : use utils::lsn::Lsn;
      17              : 
      18              : use crate::control_file;
      19              : use crate::safekeeper::{AcceptorState, PgUuid, TermHistory, TermLsn, UNKNOWN_SERVER_VERSION};
      20              : use crate::timeline::TimelineError;
      21              : use crate::wal_backup_partial::{self};
      22              : 
      23              : /// Persistent information stored on safekeeper node about timeline.
      24              : /// On disk data is prefixed by magic and format version and followed by checksum.
      25           12 : #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
      26              : pub struct TimelinePersistentState {
      27              :     #[serde(with = "hex")]
      28              :     pub tenant_id: TenantId,
      29              :     #[serde(with = "hex")]
      30              :     pub timeline_id: TimelineId,
      31              :     /// Membership configuration.
      32              :     pub mconf: Configuration,
      33              :     /// persistent acceptor state
      34              :     pub acceptor_state: AcceptorState,
      35              :     /// information about server
      36              :     pub server: ServerInfo,
      37              :     /// Unique id of the last *elected* proposer we dealt with. Not needed
      38              :     /// for correctness, exists for monitoring purposes.
      39              :     #[serde(with = "hex")]
      40              :     pub proposer_uuid: PgUuid,
      41              :     /// Since which LSN this timeline generally starts. Safekeeper might have
      42              :     /// joined later.
      43              :     pub timeline_start_lsn: Lsn,
      44              :     /// Since which LSN safekeeper has (had) WAL for this timeline.
      45              :     /// All WAL segments next to one containing local_start_lsn are
      46              :     /// filled with data from the beginning.
      47              :     pub local_start_lsn: Lsn,
      48              :     /// Part of WAL acknowledged by quorum *and available locally*. Always points
      49              :     /// to record boundary.
      50              :     pub commit_lsn: Lsn,
      51              :     /// LSN that points to the end of the last backed up segment. Useful to
      52              :     /// persist to avoid finding out offloading progress on boot.
      53              :     pub backup_lsn: Lsn,
      54              :     /// Minimal LSN which may be needed for recovery of some safekeeper (end_lsn
      55              :     /// of last record streamed to everyone). Persisting it helps skipping
      56              :     /// recovery in walproposer, generally we compute it from peers. In
      57              :     /// walproposer proto called 'truncate_lsn'. Updates are currently drived
      58              :     /// only by walproposer.
      59              :     pub peer_horizon_lsn: Lsn,
      60              :     /// LSN of the oldest known checkpoint made by pageserver and successfully
      61              :     /// pushed to s3. We don't remove WAL beyond it. Persisted only for
      62              :     /// informational purposes, we receive it from pageserver (or broker).
      63              :     pub remote_consistent_lsn: Lsn,
      64              :     /// Holds names of partial segments uploaded to remote storage. Used to
      65              :     /// clean up old objects without leaving garbage in remote storage.
      66              :     pub partial_backup: wal_backup_partial::State,
      67              :     /// Eviction state of the timeline. If it's Offloaded, we should download
      68              :     /// WAL files from remote storage to serve the timeline.
      69              :     pub eviction_state: EvictionState,
      70              :     pub creation_ts: SystemTime,
      71              : }
      72              : 
      73              : /// State of the local WAL files. Used to track current timeline state,
      74              : /// that can be either WAL files are present on disk or last partial segment
      75              : /// is offloaded to remote storage.
      76            2 : #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
      77              : pub enum EvictionState {
      78              :     /// WAL files are present on disk.
      79              :     Present,
      80              :     /// Last partial segment is offloaded to remote storage.
      81              :     /// Contains flush_lsn of the last offloaded segment.
      82              :     Offloaded(Lsn),
      83              : }
      84              : 
      85              : impl TimelinePersistentState {
      86              :     /// commit_lsn is the same as start_lsn in the normal creaiton; see
      87              :     /// `TimelineCreateRequest` comments.`
      88         1434 :     pub fn new(
      89         1434 :         ttid: &TenantTimelineId,
      90         1434 :         mconf: Configuration,
      91         1434 :         server_info: ServerInfo,
      92         1434 :         start_lsn: Lsn,
      93         1434 :         commit_lsn: Lsn,
      94         1434 :     ) -> anyhow::Result<TimelinePersistentState> {
      95         1434 :         if server_info.wal_seg_size == 0 {
      96            0 :             bail!(TimelineError::UninitializedWalSegSize(*ttid));
      97         1434 :         }
      98         1434 : 
      99         1434 :         if server_info.pg_version == UNKNOWN_SERVER_VERSION {
     100            0 :             bail!(TimelineError::UninitialinzedPgVersion(*ttid));
     101         1434 :         }
     102         1434 : 
     103         1434 :         if commit_lsn < start_lsn {
     104            0 :             bail!(
     105            0 :                 "commit_lsn {} is smaller than start_lsn {}",
     106            0 :                 commit_lsn,
     107            0 :                 start_lsn
     108            0 :             );
     109         1434 :         }
     110              : 
     111              :         // If we are given with init LSN, initialize term history with it. It
     112              :         // ensures that walproposer always must be able to find a common point
     113              :         // in histories; if it can't something is corrupted. Not having LSN here
     114              :         // is so far left for legacy case where timeline is created by compute
     115              :         // and LSN during creation is not known yet.
     116         1434 :         let term_history = if commit_lsn != Lsn::INVALID {
     117            0 :             TermHistory(vec![TermLsn {
     118            0 :                 term: INITIAL_TERM,
     119            0 :                 lsn: start_lsn,
     120            0 :             }])
     121              :         } else {
     122         1434 :             TermHistory::empty()
     123              :         };
     124              : 
     125         1434 :         Ok(TimelinePersistentState {
     126         1434 :             tenant_id: ttid.tenant_id,
     127         1434 :             timeline_id: ttid.timeline_id,
     128         1434 :             mconf,
     129         1434 :             acceptor_state: AcceptorState {
     130         1434 :                 term: INITIAL_TERM,
     131         1434 :                 term_history,
     132         1434 :             },
     133         1434 :             server: server_info,
     134         1434 :             proposer_uuid: [0; 16],
     135         1434 :             timeline_start_lsn: start_lsn,
     136         1434 :             local_start_lsn: start_lsn,
     137         1434 :             commit_lsn,
     138         1434 :             backup_lsn: start_lsn,
     139         1434 :             peer_horizon_lsn: start_lsn,
     140         1434 :             remote_consistent_lsn: Lsn(0),
     141         1434 :             partial_backup: wal_backup_partial::State::default(),
     142         1434 :             eviction_state: EvictionState::Present,
     143         1434 :             creation_ts: SystemTime::now(),
     144         1434 :         })
     145         1434 :     }
     146              : 
     147           10 :     pub fn empty() -> Self {
     148           10 :         TimelinePersistentState::new(
     149           10 :             &TenantTimelineId::empty(),
     150           10 :             Configuration::empty(),
     151           10 :             ServerInfo {
     152           10 :                 pg_version: 170000, /* Postgres server version (major * 10000) */
     153           10 :                 system_id: 0,       /* Postgres system identifier */
     154           10 :                 wal_seg_size: WAL_SEGMENT_SIZE as u32,
     155           10 :             },
     156           10 :             Lsn::INVALID,
     157           10 :             Lsn::INVALID,
     158           10 :         )
     159           10 :         .unwrap()
     160           10 :     }
     161              : }
     162              : 
     163            0 : #[derive(Debug, Clone, Serialize, Deserialize)]
     164              : // In memory safekeeper state. Fields mirror ones in `SafeKeeperPersistentState`; values
     165              : // are not flushed yet.
     166              : pub struct TimelineMemState {
     167              :     pub commit_lsn: Lsn,
     168              :     pub backup_lsn: Lsn,
     169              :     pub peer_horizon_lsn: Lsn,
     170              :     pub remote_consistent_lsn: Lsn,
     171              :     #[serde(with = "hex")]
     172              :     pub proposer_uuid: PgUuid,
     173              : }
     174              : 
     175              : /// Safekeeper persistent state plus in memory layer.
     176              : ///
     177              : /// Allows us to avoid frequent fsyncs when we update fields like commit_lsn
     178              : /// which don't need immediate persistence. Provides transactional like API
     179              : /// to atomically update the state.
     180              : ///
     181              : /// Implements Deref into *persistent* part.
     182              : pub struct TimelineState<CTRL: control_file::Storage> {
     183              :     pub inmem: TimelineMemState,
     184              :     pub pers: CTRL, // persistent
     185              : }
     186              : 
     187              : impl<CTRL> TimelineState<CTRL>
     188              : where
     189              :     CTRL: control_file::Storage,
     190              : {
     191         8712 :     pub fn new(state: CTRL) -> Self {
     192         8712 :         TimelineState {
     193         8712 :             inmem: TimelineMemState {
     194         8712 :                 commit_lsn: state.commit_lsn,
     195         8712 :                 backup_lsn: state.backup_lsn,
     196         8712 :                 peer_horizon_lsn: state.peer_horizon_lsn,
     197         8712 :                 remote_consistent_lsn: state.remote_consistent_lsn,
     198         8712 :                 proposer_uuid: state.proposer_uuid,
     199         8712 :             },
     200         8712 :             pers: state,
     201         8712 :         }
     202         8712 :     }
     203              : 
     204              :     /// Start atomic change. Returns SafeKeeperPersistentState with in memory
     205              :     /// values applied; the protocol is to 1) change returned struct as desired
     206              :     /// 2) atomically persist it with finish_change.
     207         4004 :     pub fn start_change(&self) -> TimelinePersistentState {
     208         4004 :         let mut s = self.pers.clone();
     209         4004 :         s.commit_lsn = self.inmem.commit_lsn;
     210         4004 :         s.backup_lsn = self.inmem.backup_lsn;
     211         4004 :         s.peer_horizon_lsn = self.inmem.peer_horizon_lsn;
     212         4004 :         s.remote_consistent_lsn = self.inmem.remote_consistent_lsn;
     213         4004 :         s.proposer_uuid = self.inmem.proposer_uuid;
     214         4004 :         s
     215         4004 :     }
     216              : 
     217              :     /// Persist given state. c.f. start_change.
     218         4004 :     pub async fn finish_change(&mut self, s: &TimelinePersistentState) -> Result<()> {
     219         4004 :         if s.eq(&*self.pers) {
     220           42 :             // nothing to do if state didn't change
     221           42 :         } else {
     222         3962 :             self.pers.persist(s).await?;
     223              :         }
     224              : 
     225              :         // keep in memory values up to date
     226         4004 :         self.inmem.commit_lsn = s.commit_lsn;
     227         4004 :         self.inmem.backup_lsn = s.backup_lsn;
     228         4004 :         self.inmem.peer_horizon_lsn = s.peer_horizon_lsn;
     229         4004 :         self.inmem.remote_consistent_lsn = s.remote_consistent_lsn;
     230         4004 :         self.inmem.proposer_uuid = s.proposer_uuid;
     231         4004 :         Ok(())
     232           19 :     }
     233              : 
     234              :     /// Flush in memory values.
     235          130 :     pub async fn flush(&mut self) -> Result<()> {
     236          130 :         let s = self.start_change();
     237          130 :         self.finish_change(&s).await
     238            4 :     }
     239              : 
     240              :     /// Make term at least as `to`. If `to` is None, increment current one. This
     241              :     /// is not in safekeeper.rs because we want to be able to do it even if
     242              :     /// timeline is offloaded.
     243            0 :     pub async fn term_bump(&mut self, to: Option<Term>) -> Result<TimelineTermBumpResponse> {
     244            0 :         let before = self.acceptor_state.term;
     245            0 :         let mut state = self.start_change();
     246            0 :         let new = match to {
     247            0 :             Some(to) => max(state.acceptor_state.term, to),
     248            0 :             None => state.acceptor_state.term + 1,
     249              :         };
     250            0 :         if new > state.acceptor_state.term {
     251            0 :             state.acceptor_state.term = new;
     252            0 :             self.finish_change(&state).await?;
     253            0 :         }
     254            0 :         let after = self.acceptor_state.term;
     255            0 :         Ok(TimelineTermBumpResponse {
     256            0 :             previous_term: before,
     257            0 :             current_term: after,
     258            0 :         })
     259            0 :     }
     260              : 
     261              :     /// Switch into membership configuration `to` if it is higher than the
     262              :     /// current one.
     263        19681 :     pub async fn membership_switch(
     264        19681 :         &mut self,
     265        19681 :         to: Configuration,
     266        19681 :     ) -> Result<TimelineMembershipSwitchResponse> {
     267        19681 :         let before = self.mconf.clone();
     268        19681 :         // Is switch allowed?
     269        19681 :         if to.generation <= self.mconf.generation {
     270        19681 :             info!(
     271            0 :                 "ignoring request to switch membership conf to {}, current conf {}",
     272            0 :                 to, self.mconf
     273              :             );
     274              :         } else {
     275            0 :             let mut state = self.start_change();
     276            0 :             state.mconf = to.clone();
     277            0 :             self.finish_change(&state).await?;
     278            0 :             info!("switched membership conf to {} from {}", to, before);
     279              :         }
     280        19681 :         Ok(TimelineMembershipSwitchResponse {
     281        19681 :             previous_conf: before,
     282        19681 :             current_conf: self.mconf.clone(),
     283        19681 :         })
     284            0 :     }
     285              : }
     286              : 
     287              : impl<CTRL> Deref for TimelineState<CTRL>
     288              : where
     289              :     CTRL: control_file::Storage,
     290              : {
     291              :     type Target = TimelinePersistentState;
     292              : 
     293       290427 :     fn deref(&self) -> &Self::Target {
     294       290427 :         &self.pers
     295       290427 :     }
     296              : }
        

Generated by: LCOV version 2.1-beta