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

Generated by: LCOV version 2.1-beta