LCOV - code coverage report
Current view: top level - safekeeper/tests/walproposer_sim - safekeeper.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 94.8 % 286 271
Test Date: 2025-02-20 13:11:02 Functions: 94.9 % 39 37

            Line data    Source code
       1              : //! Safekeeper communication endpoint to WAL proposer (compute node).
       2              : //! Gets messages from the network, passes them down to consensus module and
       3              : //! sends replies back.
       4              : 
       5              : use std::{collections::HashMap, sync::Arc, time::Duration};
       6              : 
       7              : use anyhow::{bail, Result};
       8              : use bytes::{Bytes, BytesMut};
       9              : use camino::Utf8PathBuf;
      10              : use desim::{
      11              :     executor::{self, PollSome},
      12              :     network::TCP,
      13              :     node_os::NodeOs,
      14              :     proto::{AnyMessage, NetEvent, NodeEvent},
      15              : };
      16              : use http::Uri;
      17              : use safekeeper::{
      18              :     safekeeper::{
      19              :         ProposerAcceptorMessage, SafeKeeper, SK_PROTOCOL_VERSION, UNKNOWN_SERVER_VERSION,
      20              :     },
      21              :     state::{TimelinePersistentState, TimelineState},
      22              :     timeline::TimelineError,
      23              :     wal_storage::Storage,
      24              :     SafeKeeperConf,
      25              : };
      26              : use safekeeper_api::{membership::Configuration, ServerInfo};
      27              : use tracing::{debug, info_span, warn};
      28              : use utils::{
      29              :     id::{NodeId, TenantId, TenantTimelineId, TimelineId},
      30              :     lsn::Lsn,
      31              : };
      32              : 
      33              : use super::safekeeper_disk::{DiskStateStorage, DiskWALStorage, SafekeeperDisk, TimelineDisk};
      34              : 
      35              : struct SharedState {
      36              :     sk: SafeKeeper<DiskStateStorage, DiskWALStorage>,
      37              :     disk: Arc<TimelineDisk>,
      38              : }
      39              : 
      40              : struct GlobalMap {
      41              :     timelines: HashMap<TenantTimelineId, SharedState>,
      42              :     conf: SafeKeeperConf,
      43              :     disk: Arc<SafekeeperDisk>,
      44              : }
      45              : 
      46              : impl GlobalMap {
      47              :     /// Restores global state from disk.
      48         9557 :     fn new(disk: Arc<SafekeeperDisk>, conf: SafeKeeperConf) -> Result<Self> {
      49         9557 :         let mut timelines = HashMap::new();
      50              : 
      51         9557 :         for (&ttid, disk) in disk.timelines.lock().iter() {
      52         7074 :             debug!("loading timeline {}", ttid);
      53         7074 :             let state = disk.state.lock().clone();
      54         7074 : 
      55         7074 :             if state.server.wal_seg_size == 0 {
      56            0 :                 bail!(TimelineError::UninitializedWalSegSize(ttid));
      57         7074 :             }
      58         7074 : 
      59         7074 :             if state.server.pg_version == UNKNOWN_SERVER_VERSION {
      60            0 :                 bail!(TimelineError::UninitialinzedPgVersion(ttid));
      61         7074 :             }
      62         7074 : 
      63         7074 :             if state.commit_lsn < state.local_start_lsn {
      64            0 :                 bail!(
      65            0 :                     "commit_lsn {} is smaller than local_start_lsn {}",
      66            0 :                     state.commit_lsn,
      67            0 :                     state.local_start_lsn
      68            0 :                 );
      69         7074 :             }
      70         7074 : 
      71         7074 :             let control_store = DiskStateStorage::new(disk.clone());
      72         7074 :             let wal_store = DiskWALStorage::new(disk.clone(), &control_store)?;
      73              : 
      74         7074 :             let sk = SafeKeeper::new(TimelineState::new(control_store), wal_store, conf.my_id)?;
      75         7074 :             timelines.insert(
      76         7074 :                 ttid,
      77         7074 :                 SharedState {
      78         7074 :                     sk,
      79         7074 :                     disk: disk.clone(),
      80         7074 :                 },
      81         7074 :             );
      82              :         }
      83              : 
      84         9557 :         Ok(Self {
      85         9557 :             timelines,
      86         9557 :             conf,
      87         9557 :             disk,
      88         9557 :         })
      89         9557 :     }
      90              : 
      91         1438 :     fn create(&mut self, ttid: TenantTimelineId, server_info: ServerInfo) -> Result<()> {
      92         1438 :         if self.timelines.contains_key(&ttid) {
      93            0 :             bail!("timeline {} already exists", ttid);
      94         1438 :         }
      95         1438 : 
      96         1438 :         debug!("creating new timeline {}", ttid);
      97              : 
      98         1438 :         let commit_lsn = Lsn::INVALID;
      99         1438 :         let local_start_lsn = Lsn::INVALID;
     100              : 
     101         1438 :         let state = TimelinePersistentState::new(
     102         1438 :             &ttid,
     103         1438 :             Configuration::empty(),
     104         1438 :             server_info,
     105         1438 :             commit_lsn,
     106         1438 :             local_start_lsn,
     107         1438 :         )?;
     108              : 
     109         1438 :         let disk_timeline = self.disk.put_state(&ttid, state);
     110         1438 :         let control_store = DiskStateStorage::new(disk_timeline.clone());
     111         1438 :         let wal_store = DiskWALStorage::new(disk_timeline.clone(), &control_store)?;
     112              : 
     113         1438 :         let sk = SafeKeeper::new(
     114         1438 :             TimelineState::new(control_store),
     115         1438 :             wal_store,
     116         1438 :             self.conf.my_id,
     117         1438 :         )?;
     118              : 
     119         1438 :         self.timelines.insert(
     120         1438 :             ttid,
     121         1438 :             SharedState {
     122         1438 :                 sk,
     123         1438 :                 disk: disk_timeline,
     124         1438 :             },
     125         1438 :         );
     126         1438 :         Ok(())
     127         1438 :     }
     128              : 
     129        26509 :     fn get(&mut self, ttid: &TenantTimelineId) -> &mut SharedState {
     130        26509 :         self.timelines.get_mut(ttid).expect("timeline must exist")
     131        26509 :     }
     132              : 
     133        17877 :     fn has_tli(&self, ttid: &TenantTimelineId) -> bool {
     134        17877 :         self.timelines.contains_key(ttid)
     135        17877 :     }
     136              : }
     137              : 
     138              : /// State of a single connection to walproposer.
     139              : struct ConnState {
     140              :     tcp: TCP,
     141              : 
     142              :     greeting: bool,
     143              :     ttid: TenantTimelineId,
     144              :     flush_pending: bool,
     145              : 
     146              :     runtime: tokio::runtime::Runtime,
     147              : }
     148              : 
     149         9557 : pub fn run_server(os: NodeOs, disk: Arc<SafekeeperDisk>) -> Result<()> {
     150         9557 :     let _enter = info_span!("safekeeper", id = os.id()).entered();
     151         9557 :     debug!("started server");
     152         9557 :     os.log_event("started;safekeeper".to_owned());
     153         9557 :     let conf = SafeKeeperConf {
     154         9557 :         workdir: Utf8PathBuf::from("."),
     155         9557 :         my_id: NodeId(os.id() as u64),
     156         9557 :         listen_pg_addr: String::new(),
     157         9557 :         listen_http_addr: String::new(),
     158         9557 :         no_sync: false,
     159         9557 :         broker_endpoint: "/".parse::<Uri>().unwrap(),
     160         9557 :         broker_keepalive_interval: Duration::from_secs(0),
     161         9557 :         heartbeat_timeout: Duration::from_secs(0),
     162         9557 :         remote_storage: None,
     163         9557 :         max_offloader_lag_bytes: 0,
     164         9557 :         wal_backup_enabled: false,
     165         9557 :         listen_pg_addr_tenant_only: None,
     166         9557 :         advertise_pg_addr: None,
     167         9557 :         availability_zone: None,
     168         9557 :         peer_recovery_enabled: false,
     169         9557 :         backup_parallel_jobs: 0,
     170         9557 :         pg_auth: None,
     171         9557 :         pg_tenant_only_auth: None,
     172         9557 :         http_auth: None,
     173         9557 :         sk_auth_token: None,
     174         9557 :         current_thread_runtime: false,
     175         9557 :         walsenders_keep_horizon: false,
     176         9557 :         partial_backup_timeout: Duration::from_secs(0),
     177         9557 :         disable_periodic_broker_push: false,
     178         9557 :         enable_offload: false,
     179         9557 :         delete_offloaded_wal: false,
     180         9557 :         control_file_save_interval: Duration::from_secs(1),
     181         9557 :         partial_backup_concurrency: 1,
     182         9557 :         eviction_min_resident: Duration::ZERO,
     183         9557 :         wal_reader_fanout: false,
     184         9557 :         max_delta_for_fanout: None,
     185         9557 :     };
     186              : 
     187         9557 :     let mut global = GlobalMap::new(disk, conf.clone())?;
     188         9557 :     let mut conns: HashMap<usize, ConnState> = HashMap::new();
     189              : 
     190         9557 :     for (&_ttid, shared_state) in global.timelines.iter_mut() {
     191         7074 :         let flush_lsn = shared_state.sk.wal_store.flush_lsn();
     192         7074 :         let commit_lsn = shared_state.sk.state.commit_lsn;
     193         7074 :         os.log_event(format!("tli_loaded;{};{}", flush_lsn.0, commit_lsn.0));
     194         7074 :     }
     195              : 
     196         9557 :     let node_events = os.node_events();
     197         9557 :     let mut epoll_vec: Vec<Box<dyn PollSome>> = vec![];
     198         9557 :     let mut epoll_idx: Vec<usize> = vec![];
     199              : 
     200              :     // TODO: batch events processing (multiple events per tick)
     201              :     loop {
     202        68534 :         epoll_vec.clear();
     203        68534 :         epoll_idx.clear();
     204        68534 : 
     205        68534 :         // node events channel
     206        68534 :         epoll_vec.push(Box::new(node_events.clone()));
     207        68534 :         epoll_idx.push(0);
     208              : 
     209              :         // tcp connections
     210       226352 :         for conn in conns.values() {
     211       226352 :             epoll_vec.push(Box::new(conn.tcp.recv_chan()));
     212       226352 :             epoll_idx.push(conn.tcp.connection_id());
     213       226352 :         }
     214              : 
     215              :         // waiting for the next message
     216        68534 :         let index = executor::epoll_chans(&epoll_vec, -1).unwrap();
     217        68534 : 
     218        68534 :         if index == 0 {
     219              :             // got a new connection
     220        23060 :             match node_events.must_recv() {
     221        23060 :                 NodeEvent::Accept(tcp) => {
     222        23060 :                     conns.insert(
     223        23060 :                         tcp.connection_id(),
     224        23060 :                         ConnState {
     225        23060 :                             tcp,
     226        23060 :                             greeting: false,
     227        23060 :                             ttid: TenantTimelineId::empty(),
     228        23060 :                             flush_pending: false,
     229        23060 :                             runtime: tokio::runtime::Builder::new_current_thread().build()?,
     230              :                         },
     231              :                     );
     232              :                 }
     233            0 :                 NodeEvent::Internal(_) => unreachable!(),
     234              :             }
     235        23060 :             continue;
     236        45474 :         }
     237        45474 : 
     238        45474 :         let connection_id = epoll_idx[index];
     239        45474 :         let conn = conns.get_mut(&connection_id).unwrap();
     240        45474 :         let mut next_event = Some(conn.tcp.recv_chan().must_recv());
     241              : 
     242              :         loop {
     243        82006 :             let event = match next_event {
     244        46322 :                 Some(event) => event,
     245        35684 :                 None => break,
     246              :             };
     247              : 
     248        46322 :             match event {
     249        33740 :                 NetEvent::Message(msg) => {
     250        33740 :                     let res = conn.process_any(msg, &mut global);
     251        33740 :                     if res.is_err() {
     252         9790 :                         let e = res.unwrap_err();
     253         9790 :                         let estr = e.to_string();
     254         9790 :                         if !estr.contains("finished processing START_REPLICATION") {
     255         9557 :                             warn!("conn {:?} error: {:?}", connection_id, e);
     256            0 :                             panic!("unexpected error at safekeeper: {:#}", e);
     257          233 :                         }
     258          233 :                         conns.remove(&connection_id);
     259          233 :                         break;
     260        23950 :                     }
     261              :                 }
     262        12582 :                 NetEvent::Closed => {
     263        12582 :                     // TODO: remove from conns?
     264        12582 :                 }
     265              :             }
     266              : 
     267        36532 :             next_event = conn.tcp.recv_chan().try_recv();
     268              :         }
     269              : 
     270       145834 :         conns.retain(|_, conn| {
     271       145834 :             let res = conn.flush(&mut global);
     272       145834 :             if res.is_err() {
     273            0 :                 debug!("conn {:?} error: {:?}", conn.tcp, res);
     274       145834 :             }
     275       145834 :             res.is_ok()
     276       145834 :         });
     277              :     }
     278            0 : }
     279              : 
     280              : impl ConnState {
     281              :     /// Process a message from the network. It can be START_REPLICATION request or a valid ProposerAcceptorMessage message.
     282        24183 :     fn process_any(&mut self, any: AnyMessage, global: &mut GlobalMap) -> Result<()> {
     283        24183 :         if let AnyMessage::Bytes(copy_data) = any {
     284        24183 :             let repl_prefix = b"START_REPLICATION ";
     285        24183 :             if !self.greeting && copy_data.starts_with(repl_prefix) {
     286          233 :                 self.process_start_replication(copy_data.slice(repl_prefix.len()..), global)?;
     287          233 :                 bail!("finished processing START_REPLICATION")
     288        23950 :             }
     289              : 
     290        23950 :             let msg = ProposerAcceptorMessage::parse(copy_data, SK_PROTOCOL_VERSION)?;
     291        23950 :             debug!("got msg: {:?}", msg);
     292        23950 :             self.process(msg, global)
     293              :         } else {
     294            0 :             bail!("unexpected message, expected AnyMessage::Bytes");
     295              :         }
     296        24183 :     }
     297              : 
     298              :     /// Process START_REPLICATION request.
     299          233 :     fn process_start_replication(
     300          233 :         &mut self,
     301          233 :         copy_data: Bytes,
     302          233 :         global: &mut GlobalMap,
     303          233 :     ) -> Result<()> {
     304              :         // format is "<tenant_id> <timeline_id> <start_lsn> <end_lsn>"
     305          233 :         let str = String::from_utf8(copy_data.to_vec())?;
     306              : 
     307          233 :         let mut parts = str.split(' ');
     308          233 :         let tenant_id = parts.next().unwrap().parse::<TenantId>()?;
     309          233 :         let timeline_id = parts.next().unwrap().parse::<TimelineId>()?;
     310          233 :         let start_lsn = parts.next().unwrap().parse::<u64>()?;
     311          233 :         let end_lsn = parts.next().unwrap().parse::<u64>()?;
     312              : 
     313          233 :         let ttid = TenantTimelineId::new(tenant_id, timeline_id);
     314          233 :         let shared_state = global.get(&ttid);
     315          233 : 
     316          233 :         // read bytes from start_lsn to end_lsn
     317          233 :         let mut buf = vec![0; (end_lsn - start_lsn) as usize];
     318          233 :         shared_state.disk.wal.lock().read(start_lsn, &mut buf);
     319          233 : 
     320          233 :         // send bytes to the client
     321          233 :         self.tcp.send(AnyMessage::Bytes(Bytes::from(buf)));
     322          233 :         Ok(())
     323          233 :     }
     324              : 
     325              :     /// Get or create a timeline.
     326        17877 :     fn init_timeline(
     327        17877 :         &mut self,
     328        17877 :         ttid: TenantTimelineId,
     329        17877 :         server_info: ServerInfo,
     330        17877 :         global: &mut GlobalMap,
     331        17877 :     ) -> Result<()> {
     332        17877 :         self.ttid = ttid;
     333        17877 :         if global.has_tli(&ttid) {
     334        16439 :             return Ok(());
     335         1438 :         }
     336         1438 : 
     337         1438 :         global.create(ttid, server_info)
     338        17877 :     }
     339              : 
     340              :     /// Process a ProposerAcceptorMessage.
     341        23950 :     fn process(&mut self, msg: ProposerAcceptorMessage, global: &mut GlobalMap) -> Result<()> {
     342        23950 :         if !self.greeting {
     343        17877 :             self.greeting = true;
     344        17877 : 
     345        17877 :             match msg {
     346        17877 :                 ProposerAcceptorMessage::Greeting(ref greeting) => {
     347        17877 :                     tracing::info!(
     348            0 :                         "start handshake with walproposer {:?} {:?}",
     349              :                         self.tcp,
     350              :                         greeting
     351              :                     );
     352        17877 :                     let server_info = ServerInfo {
     353        17877 :                         pg_version: greeting.pg_version,
     354        17877 :                         system_id: greeting.system_id,
     355        17877 :                         wal_seg_size: greeting.wal_seg_size,
     356        17877 :                     };
     357        17877 :                     let ttid = TenantTimelineId::new(greeting.tenant_id, greeting.timeline_id);
     358        17877 :                     self.init_timeline(ttid, server_info, global)?
     359              :                 }
     360              :                 _ => {
     361            0 :                     bail!("unexpected message {msg:?} instead of greeting");
     362              :                 }
     363              :             }
     364         6073 :         }
     365              : 
     366        23950 :         let tli = global.get(&self.ttid);
     367        23950 : 
     368        23950 :         match msg {
     369         2820 :             ProposerAcceptorMessage::AppendRequest(append_request) => {
     370         2820 :                 self.flush_pending = true;
     371         2820 :                 self.process_sk_msg(
     372         2820 :                     tli,
     373         2820 :                     &ProposerAcceptorMessage::NoFlushAppendRequest(append_request),
     374         2820 :                 )?;
     375              :             }
     376        21130 :             other => {
     377        21130 :                 self.process_sk_msg(tli, &other)?;
     378              :             }
     379              :         }
     380              : 
     381        23950 :         Ok(())
     382        23950 :     }
     383              : 
     384              :     /// Process FlushWAL if needed.
     385       145834 :     fn flush(&mut self, global: &mut GlobalMap) -> Result<()> {
     386       145834 :         // TODO: try to add extra flushes in simulation, to verify that extra flushes don't break anything
     387       145834 :         if !self.flush_pending {
     388       143508 :             return Ok(());
     389         2326 :         }
     390         2326 :         self.flush_pending = false;
     391         2326 :         let shared_state = global.get(&self.ttid);
     392         2326 :         self.process_sk_msg(shared_state, &ProposerAcceptorMessage::FlushWAL)
     393       145834 :     }
     394              : 
     395              :     /// Make safekeeper process a message and send a reply to the TCP
     396        26276 :     fn process_sk_msg(
     397        26276 :         &mut self,
     398        26276 :         shared_state: &mut SharedState,
     399        26276 :         msg: &ProposerAcceptorMessage,
     400        26276 :     ) -> Result<()> {
     401        26276 :         let mut reply = self.runtime.block_on(shared_state.sk.process_msg(msg))?;
     402        26276 :         if let Some(reply) = &mut reply {
     403              :             // TODO: if this is AppendResponse, fill in proper hot standby feedback and disk consistent lsn
     404              : 
     405        22813 :             let mut buf = BytesMut::with_capacity(128);
     406        22813 :             reply.serialize(&mut buf)?;
     407              : 
     408        22813 :             self.tcp.send(AnyMessage::Bytes(buf.into()));
     409         3463 :         }
     410        26276 :         Ok(())
     411        26276 :     }
     412              : }
     413              : 
     414              : impl Drop for ConnState {
     415        23032 :     fn drop(&mut self) {
     416        23032 :         debug!("dropping conn: {:?}", self.tcp);
     417        23032 :         if !std::thread::panicking() {
     418          233 :             self.tcp.close();
     419        22799 :         }
     420              :         // TODO: clean up non-fsynced WAL
     421        23032 :     }
     422              : }
        

Generated by: LCOV version 2.1-beta