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

Generated by: LCOV version 2.1-beta