From 0faf0e92ec5ad897b2421d39d1085a9e2890c86d Mon Sep 17 00:00:00 2001 From: Arthur Petukhovsky Date: Tue, 2 Aug 2022 11:29:23 +0000 Subject: [PATCH] Add TimelineState --- Cargo.lock | 2 + safekeeper/Cargo.toml | 2 + safekeeper/src/bin/safekeeper.rs | 4 +- safekeeper/src/broker.rs | 37 +- safekeeper/src/control_file.rs | 8 - safekeeper/src/handler.rs | 33 +- safekeeper/src/http/routes.rs | 21 +- safekeeper/src/json_ctrl.rs | 31 +- safekeeper/src/lib.rs | 3 + safekeeper/src/metrics.rs | 13 +- safekeeper/src/receive_wal.rs | 9 +- safekeeper/src/remove_wal.rs | 21 +- safekeeper/src/safekeeper.rs | 135 +++-- safekeeper/src/send_wal.rs | 33 +- safekeeper/src/timeline.rs | 730 +++++++++++++------------ safekeeper/src/timelines_global_map.rs | 72 +++ safekeeper/src/wal_backup.rs | 41 +- safekeeper/src/wal_storage.rs | 241 ++++---- 18 files changed, 798 insertions(+), 638 deletions(-) create mode 100644 safekeeper/src/timelines_global_map.rs diff --git a/Cargo.lock b/Cargo.lock index e9ebcdc5ac..e6d0b6be25 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2721,6 +2721,7 @@ dependencies = [ "hyper", "metrics", "once_cell", + "parking_lot 0.12.1", "postgres", "postgres-protocol", "postgres_ffi", @@ -2731,6 +2732,7 @@ dependencies = [ "serde_with", "signal-hook", "tempfile", + "thiserror", "tokio", "tokio-postgres", "toml_edit", diff --git a/safekeeper/Cargo.toml b/safekeeper/Cargo.toml index 4ed30413e2..c25772e7b9 100644 --- a/safekeeper/Cargo.toml +++ b/safekeeper/Cargo.toml @@ -30,6 +30,8 @@ git-version = "0.3.5" async-trait = "0.1" once_cell = "1.13.0" toml_edit = { version = "0.13", features = ["easy"] } +thiserror = "1" +parking_lot = "0.12.1" postgres_ffi = { path = "../libs/postgres_ffi" } metrics = { path = "../libs/metrics" } diff --git a/safekeeper/src/bin/safekeeper.rs b/safekeeper/src/bin/safekeeper.rs index 244c793250..edb1af20ea 100644 --- a/safekeeper/src/bin/safekeeper.rs +++ b/safekeeper/src/bin/safekeeper.rs @@ -24,9 +24,9 @@ use safekeeper::defaults::{ }; use safekeeper::http; use safekeeper::remove_wal; -use safekeeper::timeline::GlobalTimelines; use safekeeper::wal_backup; use safekeeper::wal_service; +use safekeeper::GlobalTimelines; use safekeeper::SafeKeeperConf; use utils::auth::JwtAuth; use utils::{ @@ -298,7 +298,7 @@ fn start_safekeeper(mut conf: SafeKeeperConf, given_id: Option, init: bo let signals = signals::install_shutdown_handlers()?; let mut threads = vec![]; let (wal_backup_launcher_tx, wal_backup_launcher_rx) = mpsc::channel(100); - GlobalTimelines::init(wal_backup_launcher_tx); + GlobalTimelines::init(conf.clone(), wal_backup_launcher_tx); let conf_ = conf.clone(); threads.push( diff --git a/safekeeper/src/broker.rs b/safekeeper/src/broker.rs index ce66131700..17a96a5bcc 100644 --- a/safekeeper/src/broker.rs +++ b/safekeeper/src/broker.rs @@ -10,6 +10,7 @@ use etcd_broker::LeaseKeeper; use std::collections::hash_map::Entry; use std::collections::HashMap; +use std::collections::HashSet; use std::time::Duration; use tokio::spawn; use tokio::task::JoinHandle; @@ -17,7 +18,8 @@ use tokio::{runtime, time::sleep}; use tracing::*; use url::Url; -use crate::{timeline::GlobalTimelines, SafeKeeperConf}; +use crate::GlobalTimelines; +use crate::SafeKeeperConf; use etcd_broker::{ subscription_key::{OperationKind, SkOperationKind, SubscriptionKey}, Client, PutOptions, @@ -210,11 +212,15 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> { // is under plain mutex. That's ok, all this code is not performance // sensitive and there is no risk of deadlock as we don't await while // lock is held. - let active_tlis = GlobalTimelines::get_active_timelines(); + let mut active_tlis = GlobalTimelines::get_all(); + active_tlis.retain(|tli| tli.is_active()); + + let active_tlis_set: HashSet = + active_tlis.iter().map(|tli| tli.zttid).collect(); // // Get and maintain (if not yet) per timeline lease to automatically delete obsolete data. - for zttid in active_tlis.iter() { - if let Entry::Vacant(v) = leases.entry(*zttid) { + for tli in &active_tlis { + if let Entry::Vacant(v) = leases.entry(tli.zttid) { let lease = client.lease_grant(LEASE_TTL_SEC, None).await?; let (keeper, ka_stream) = client.lease_keep_alive(lease.id()).await?; v.insert(Lease { @@ -224,21 +230,26 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> { }); } } - leases.retain(|zttid, _| active_tlis.contains(zttid)); + leases.retain(|zttid, _| active_tlis_set.contains(zttid)); // Push data concurrently to not suffer from latency, with many timelines it can be slow. let handles = active_tlis .iter() - .filter_map(|zttid| GlobalTimelines::get_loaded(*zttid)) - .map(|tli| { - let sk_info = tli.get_public_info(&conf); + .filter_map(|tli| { + let sk_info = tli.get_public_info(&conf).ok()?; let key = timeline_safekeeper_path( conf.broker_etcd_prefix.clone(), tli.zttid, conf.my_id, ); let lease = leases.remove(&tli.zttid).unwrap(); - tokio::spawn(push_sk_info(tli.zttid, client.clone(), key, sk_info, lease)) + Some(tokio::spawn(push_sk_info( + tli.zttid, + client.clone(), + key, + sk_info, + lease, + ))) }) .collect::>(); for h in handles { @@ -279,10 +290,10 @@ async fn pull_loop(conf: SafeKeeperConf) -> Result<()> { match subscription.value_updates.recv().await { Some(new_info) => { // note: there are blocking operations below, but it's considered fine for now - if let Ok(tli) = GlobalTimelines::get(&conf, new_info.key.id, false) { - tli.record_safekeeper_info(&new_info.value, new_info.key.node_id) - .await? - } + let tli = GlobalTimelines::get(new_info.key.id); + let _ = tli + .record_safekeeper_info(&new_info.value, new_info.key.node_id) + .await; } None => { // XXX it means we lost connection with etcd, error is consumed inside sub object diff --git a/safekeeper/src/control_file.rs b/safekeeper/src/control_file.rs index 7fc75246e1..5017104dbd 100644 --- a/safekeeper/src/control_file.rs +++ b/safekeeper/src/control_file.rs @@ -9,8 +9,6 @@ use std::io::{Read, Write}; use std::ops::Deref; use std::path::{Path, PathBuf}; -use tracing::*; - use crate::control_file_upgrade::upgrade_control_file; use crate::safekeeper::{SafeKeeperState, SK_FORMAT_VERSION, SK_MAGIC}; use metrics::{register_histogram_vec, Histogram, HistogramVec, DISK_WRITE_SECONDS_BUCKETS}; @@ -122,13 +120,7 @@ impl FileStorage { } /// Read in the control file. - /// If create=false and file doesn't exist, bails out. pub fn load_control_file>(control_file_path: P) -> Result { - info!( - "loading control file {}", - control_file_path.as_ref().display(), - ); - let mut control_file = OpenOptions::new() .read(true) .write(true) diff --git a/safekeeper/src/handler.rs b/safekeeper/src/handler.rs index 3e301259ed..c64ed84e3b 100644 --- a/safekeeper/src/handler.rs +++ b/safekeeper/src/handler.rs @@ -5,8 +5,8 @@ use crate::json_ctrl::{handle_json_ctrl, AppendLogicalMessage}; use crate::receive_wal::ReceiveWalConn; use crate::safekeeper::{AcceptorProposerMessage, ProposerAcceptorMessage}; use crate::send_wal::ReplicationConn; -use crate::timeline::{Timeline, TimelineTools}; -use crate::SafeKeeperConf; +use crate::timeline::Timeline; +use crate::{GlobalTimelines, SafeKeeperConf}; use anyhow::{bail, Context, Result}; use postgres_ffi::PG_TLI; @@ -98,17 +98,12 @@ impl postgres_backend::Handler for SafekeeperPostgresHandler { query_string, self.ztimelineid ); - let create = !(matches!(cmd, SafekeeperPostgresCommand::StartReplication { .. }) - || matches!(cmd, SafekeeperPostgresCommand::IdentifySystem)); - let tenantid = self.ztenantid.context("tenantid is required")?; let timelineid = self.ztimelineid.context("timelineid is required")?; if self.timeline.is_none() { - self.timeline.set( - &self.conf, - ZTenantTimelineId::new(tenantid, timelineid), - create, - )?; + self.timeline = Some(GlobalTimelines::get(ZTenantTimelineId::new( + tenantid, timelineid, + ))); } match cmd { @@ -144,7 +139,8 @@ impl SafekeeperPostgresHandler { msg: &ProposerAcceptorMessage, ) -> Result> { self.timeline - .get() + .as_ref() + .unwrap() .process_msg(msg) .context("failed to process ProposerAcceptorMessage") } @@ -153,23 +149,18 @@ impl SafekeeperPostgresHandler { /// Handle IDENTIFY_SYSTEM replication command /// fn handle_identify_system(&mut self, pgb: &mut PostgresBackend) -> Result<()> { + let tli = self.timeline.as_ref().unwrap(); + let lsn = if self.is_walproposer_recovery() { // walproposer should get all local WAL until flush_lsn - self.timeline.get().get_end_of_wal() + tli.get_flush_lsn()? } else { // other clients shouldn't get any uncommitted WAL - self.timeline.get().get_state().0.commit_lsn + tli.get_state()?.0.commit_lsn } .to_string(); - let sysid = self - .timeline - .get() - .get_state() - .1 - .server - .system_id - .to_string(); + let sysid = tli.get_state()?.1.server.system_id.to_string(); let lsn_bytes = lsn.as_bytes(); let tli = PG_TLI.to_string(); let tli_bytes = tli.as_bytes(); diff --git a/safekeeper/src/http/routes.rs b/safekeeper/src/http/routes.rs index 13356c5921..5d843585d1 100644 --- a/safekeeper/src/http/routes.rs +++ b/safekeeper/src/http/routes.rs @@ -1,3 +1,4 @@ +use anyhow::anyhow; use hyper::{Body, Request, Response, StatusCode, Uri}; use once_cell::sync::Lazy; @@ -9,7 +10,9 @@ use std::sync::Arc; use crate::safekeeper::Term; use crate::safekeeper::TermHistory; -use crate::timeline::{GlobalTimelines, TimelineDeleteForceResult}; +use crate::timeline; +use crate::timeline::TimelineDeleteForceResult; +use crate::GlobalTimelines; use crate::SafeKeeperConf; use etcd_broker::subscription_value::SkTimelineInfo; use utils::{ @@ -96,9 +99,9 @@ async fn timeline_status_handler(request: Request) -> Result) -> Result) -> Result Result<()> { wal_seg_size: WAL_SEGMENT_SIZE as u32, // 16MB, default for tests }); - let response = spg.timeline.get().process_msg(&greeting_request)?; + let response = spg + .timeline + .as_ref() + .unwrap() + .process_msg(&greeting_request)?; match response { Some(AcceptorProposerMessage::Greeting(_)) => Ok(()), _ => anyhow::bail!("not GreetingResponse"), @@ -112,7 +117,14 @@ fn prepare_safekeeper(spg: &mut SafekeeperPostgresHandler) -> Result<()> { fn send_proposer_elected(spg: &mut SafekeeperPostgresHandler, term: Term, lsn: Lsn) -> Result<()> { // add new term to existing history - let history = spg.timeline.get().get_state().1.acceptor_state.term_history; + let history = spg + .timeline + .as_ref() + .unwrap() + .get_state()? + .1 + .acceptor_state + .term_history; let history = history.up_to(lsn.checked_sub(1u64).unwrap()); let mut history_entries = history.0; history_entries.push(TermSwitchEntry { term, lsn }); @@ -125,7 +137,10 @@ fn send_proposer_elected(spg: &mut SafekeeperPostgresHandler, term: Term, lsn: L timeline_start_lsn: lsn, }); - spg.timeline.get().process_msg(&proposer_elected_request)?; + spg.timeline + .as_ref() + .unwrap() + .process_msg(&proposer_elected_request)?; Ok(()) } @@ -143,7 +158,7 @@ fn append_logical_message( msg: &AppendLogicalMessage, ) -> Result { let wal_data = xlog_utils::encode_logical_message(&msg.lm_prefix, &msg.lm_message); - let sk_state = spg.timeline.get().get_state().1; + let sk_state = spg.timeline.as_ref().unwrap().get_state()?.1; let begin_lsn = msg.begin_lsn; let end_lsn = begin_lsn + wal_data.len() as u64; @@ -167,7 +182,11 @@ fn append_logical_message( wal_data: Bytes::from(wal_data), }); - let response = spg.timeline.get().process_msg(&append_request)?; + let response = spg + .timeline + .as_ref() + .unwrap() + .process_msg(&append_request)?; let append_response = match response { Some(AcceptorProposerMessage::AppendResponse(resp)) => resp, diff --git a/safekeeper/src/lib.rs b/safekeeper/src/lib.rs index 0335d61d3f..5fdd5a72f0 100644 --- a/safekeeper/src/lib.rs +++ b/safekeeper/src/lib.rs @@ -23,6 +23,9 @@ pub mod wal_backup; pub mod wal_service; pub mod wal_storage; +mod timelines_global_map; +pub use timelines_global_map::GlobalTimelines; + pub mod defaults { use const_format::formatcp; use std::time::Duration; diff --git a/safekeeper/src/metrics.rs b/safekeeper/src/metrics.rs index c693035dd3..86d913c24d 100644 --- a/safekeeper/src/metrics.rs +++ b/safekeeper/src/metrics.rs @@ -12,7 +12,8 @@ use utils::{lsn::Lsn, zid::ZTenantTimelineId}; use crate::{ safekeeper::{SafeKeeperState, SafekeeperMemState}, - timeline::{GlobalTimelines, ReplicaState}, + timeline::ReplicaState, + GlobalTimelines, }; pub struct FullTimelineInfo { @@ -235,9 +236,15 @@ impl Collector for TimelineCollector { self.disk_usage.reset(); self.acceptor_term.reset(); - let timelines = GlobalTimelines::active_timelines_metrics(); + let timelines = GlobalTimelines::get_all(); + + for arc_tli in timelines { + let tli = arc_tli.info_for_metrics(); + if tli.is_none() { + continue; + } + let tli = tli.unwrap(); - for tli in timelines { let tenant_id = tli.zttid.tenant_id.to_string(); let timeline_id = tli.zttid.timeline_id.to_string(); let labels = &[tenant_id.as_str(), timeline_id.as_str()]; diff --git a/safekeeper/src/receive_wal.rs b/safekeeper/src/receive_wal.rs index af4cfb6ba4..9f3daa55d2 100644 --- a/safekeeper/src/receive_wal.rs +++ b/safekeeper/src/receive_wal.rs @@ -20,7 +20,6 @@ use crate::safekeeper::AcceptorProposerMessage; use crate::safekeeper::ProposerAcceptorMessage; use crate::handler::SafekeeperPostgresHandler; -use crate::timeline::TimelineTools; use utils::{ postgres_backend::PostgresBackend, pq_proto::{BeMessage, FeMessage}, @@ -112,9 +111,9 @@ impl<'pg> ReceiveWalConn<'pg> { // Register the connection and defer unregister. Do that only // after processing first message, as it sets wal_seg_size, // wanted by many. - spg.timeline.get().on_compute_connect()?; + spg.timeline.as_ref().unwrap().on_compute_connect()?; _guard = Some(ComputeConnectionGuard { - timeline: Arc::clone(spg.timeline.get()), + timeline: Arc::clone(spg.timeline.as_ref().unwrap()), }); first_time_through = false; } @@ -190,6 +189,8 @@ struct ComputeConnectionGuard { impl Drop for ComputeConnectionGuard { fn drop(&mut self) { - self.timeline.on_compute_disconnect().unwrap(); + if let Err(e) = self.timeline.on_compute_disconnect() { + error!("failed to unregister compute connection: {}", e); + } } } diff --git a/safekeeper/src/remove_wal.rs b/safekeeper/src/remove_wal.rs index 004c0243f9..a05b7dc580 100644 --- a/safekeeper/src/remove_wal.rs +++ b/safekeeper/src/remove_wal.rs @@ -4,20 +4,21 @@ use std::{thread, time::Duration}; use tracing::*; -use crate::{timeline::GlobalTimelines, SafeKeeperConf}; +use crate::{GlobalTimelines, SafeKeeperConf}; pub fn thread_main(conf: SafeKeeperConf) { let wal_removal_interval = Duration::from_millis(5000); loop { - let active_tlis = GlobalTimelines::get_active_timelines(); - for zttid in &active_tlis { - if let Ok(tli) = GlobalTimelines::get(&conf, *zttid, false) { - if let Err(e) = tli.remove_old_wal(conf.wal_backup_enabled) { - warn!( - "failed to remove WAL for tenant {} timeline {}: {}", - tli.zttid.tenant_id, tli.zttid.timeline_id, e - ); - } + let tlis = GlobalTimelines::get_all(); + for tli in &tlis { + if !tli.is_active() { + continue; + } + let zttid = tli.zttid; + let _enter = + info_span!("", tenant = %zttid.tenant_id, timeline = %zttid.timeline_id).entered(); + if let Err(e) = tli.remove_old_wal(conf.wal_backup_enabled) { + warn!("failed to remove WAL: {}", e); } } thread::sleep(wal_removal_interval) diff --git a/safekeeper/src/safekeeper.rs b/safekeeper/src/safekeeper.rs index a2bdcb55e7..39858e34c9 100644 --- a/safekeeper/src/safekeeper.rs +++ b/safekeeper/src/safekeeper.rs @@ -219,7 +219,11 @@ pub struct SafekeeperMemState { } impl SafeKeeperState { - pub fn new(zttid: &ZTenantTimelineId, peers: Vec) -> SafeKeeperState { + pub fn new( + zttid: &ZTenantTimelineId, + server_info: ServerInfo, + peers: Vec, + ) -> SafeKeeperState { SafeKeeperState { tenant_id: zttid.tenant_id, timeline_id: zttid.timeline_id, @@ -227,11 +231,7 @@ impl SafeKeeperState { term: 0, term_history: TermHistory::empty(), }, - server: ServerInfo { - pg_version: UNKNOWN_SERVER_VERSION, /* Postgres server version */ - system_id: 0, /* Postgres system identifier */ - wal_seg_size: 0, - }, + server: server_info, proposer_uuid: [0; 16], timeline_start_lsn: Lsn(0), local_start_lsn: Lsn(0), @@ -245,7 +245,15 @@ impl SafeKeeperState { #[cfg(test)] pub fn empty() -> Self { - SafeKeeperState::new(&ZTenantTimelineId::empty(), vec![]) + SafeKeeperState::new( + &ZTenantTimelineId::empty(), + ServerInfo { + pg_version: UNKNOWN_SERVER_VERSION, /* Postgres server version */ + system_id: 0, /* Postgres system identifier */ + wal_seg_size: 0, + }, + vec![], + ) } } @@ -481,8 +489,12 @@ impl AcceptorProposerMessage { } } -/// SafeKeeper which consumes events (messages from compute) and provides -/// replies. +/// Safekeeper implements consensus to reliably persist WAL across nodes. +/// It controls all WAL disk writes and updates of control file. +/// +/// Currently safekeeper processes: +/// - messages from compute (proposers) and provides replies +/// - messages from etcd peers pub struct SafeKeeper { /// Maximum commit_lsn between all nodes, can be ahead of local flush_lsn. /// Note: be careful to set only if we are sure our WAL (term history) matches @@ -505,19 +517,22 @@ where CTRL: control_file::Storage, WAL: wal_storage::Storage, { - // constructor - pub fn new( - ztli: ZTimelineId, - state: CTRL, - mut wal_store: WAL, - node_id: NodeId, - ) -> Result> { - if state.timeline_id != ZTimelineId::from([0u8; 16]) && ztli != state.timeline_id { - bail!("Calling SafeKeeper::new with inconsistent ztli ({}) and SafeKeeperState.server.timeline_id ({})", ztli, state.timeline_id); + /// Accepts a control file storage containing the safekeeper state. + /// State must be initialized, i.e. contain filled `tenant_id`, `timeline_id` + /// and `server` (`wal_seg_size` inside it) fields. + pub fn new(state: CTRL, wal_store: WAL, node_id: NodeId) -> Result> { + if state.tenant_id == ZTenantId::from([0u8; 16]) + || state.timeline_id == ZTimelineId::from([0u8; 16]) + { + bail!( + "Calling SafeKeeper::new with empty tenant_id ({}) or timeline_id ({})", + state.tenant_id, + state.timeline_id + ); + } + if state.server.wal_seg_size == 0 { + bail!("Calling SafeKeeper::new with empty wal_seg_size"); } - - // initialize wal_store, if state is already initialized - wal_store.init_storage(&state)?; Ok(SafeKeeper { global_commit_lsn: state.commit_lsn, @@ -576,7 +591,7 @@ where &mut self, msg: &ProposerGreeting, ) -> Result> { - /* Check protocol compatibility */ + // Check protocol compatibility if msg.protocol_version != SK_PROTOCOL_VERSION { bail!( "incompatible protocol version {}, expected {}", @@ -584,11 +599,11 @@ where SK_PROTOCOL_VERSION ); } - /* Postgres upgrade is not treated as fatal error */ + // Postgres upgrade is not treated as fatal error if msg.pg_version != self.state.server.pg_version && self.state.server.pg_version != UNKNOWN_SERVER_VERSION { - info!( + warn!( "incompatible server version {}, expected {}", msg.pg_version, self.state.server.pg_version ); @@ -607,17 +622,25 @@ where self.state.timeline_id ); } - - // set basic info about server, if not yet - // TODO: verify that is doesn't change after - { - let mut state = self.state.clone(); - state.server.system_id = msg.system_id; - state.server.wal_seg_size = msg.wal_seg_size; - self.state.persist(&state)?; + if self.state.server.wal_seg_size != msg.wal_seg_size { + bail!( + "invalid wal_seg_size, got {}, expected {}", + msg.wal_seg_size, + self.state.server.wal_seg_size + ); } - self.wal_store.init_storage(&self.state)?; + // system_id will be updated on mismatch + if self.state.server.system_id != msg.system_id { + warn!( + "unexpected system ID arrived, got {}, expected {}", + msg.system_id, self.state.server.system_id + ); + + let mut state = self.state.clone(); + state.server.system_id = msg.system_id; + self.state.persist(&state)?; + } info!( "processed greeting from proposer {:?}, sending term {:?}", @@ -667,16 +690,6 @@ where Ok(Some(AcceptorProposerMessage::VoteResponse(resp))) } - /// Bump our term if received a note from elected proposer with higher one - fn bump_if_higher(&mut self, term: Term) -> Result<()> { - if self.state.acceptor_state.term < term { - let mut state = self.state.clone(); - state.acceptor_state.term = term; - self.state.persist(&state)?; - } - Ok(()) - } - /// Form AppendResponse from current state. fn append_response(&self) -> AppendResponse { let ar = AppendResponse { @@ -693,7 +706,12 @@ where fn handle_elected(&mut self, msg: &ProposerElected) -> Result> { info!("received ProposerElected {:?}", msg); - self.bump_if_higher(msg.term)?; + if self.state.acceptor_state.term < msg.term { + let mut state = self.state.clone(); + state.acceptor_state.term = msg.term; + self.state.persist(&state)?; + } + // If our term is higher, ignore the message (next feedback will inform the compute) if self.state.acceptor_state.term > msg.term { return Ok(None); @@ -750,7 +768,7 @@ where } /// Advance commit_lsn taking into account what we have locally - pub fn update_commit_lsn(&mut self) -> Result<()> { + fn update_commit_lsn(&mut self) -> Result<()> { let commit_lsn = min(self.global_commit_lsn, self.flush_lsn()); assert!(commit_lsn >= self.inmem.commit_lsn); @@ -920,6 +938,8 @@ where #[cfg(test)] mod tests { + use postgres_ffi::v14::pg_constants; + use super::*; use crate::wal_storage::Storage; use std::ops::Deref; @@ -944,6 +964,14 @@ mod tests { } } + fn test_sk_state() -> SafeKeeperState { + let mut state = SafeKeeperState::empty(); + state.server.wal_seg_size = pg_constants::WAL_SEGMENT_SIZE as u32; + state.tenant_id = ZTenantId::from([1u8; 16]); + state.timeline_id = ZTimelineId::from([1u8; 16]); + state + } + struct DummyWalStore { lsn: Lsn, } @@ -953,10 +981,6 @@ mod tests { self.lsn } - fn init_storage(&mut self, _state: &SafeKeeperState) -> Result<()> { - Ok(()) - } - fn write_wal(&mut self, startpos: Lsn, buf: &[u8]) -> Result<()> { self.lsn = startpos + buf.len() as u64; Ok(()) @@ -979,12 +1003,10 @@ mod tests { #[test] fn test_voting() { let storage = InMemoryState { - persisted_state: SafeKeeperState::empty(), + persisted_state: test_sk_state(), }; let wal_store = DummyWalStore { lsn: Lsn(0) }; - let ztli = ZTimelineId::from([0u8; 16]); - - let mut sk = SafeKeeper::new(ztli, storage, wal_store, NodeId(0)).unwrap(); + let mut sk = SafeKeeper::new(storage, wal_store, NodeId(0)).unwrap(); // check voting for 1 is ok let vote_request = ProposerAcceptorMessage::VoteRequest(VoteRequest { term: 1 }); @@ -1000,7 +1022,7 @@ mod tests { persisted_state: state, }; - sk = SafeKeeper::new(ztli, storage, sk.wal_store, NodeId(0)).unwrap(); + sk = SafeKeeper::new(storage, sk.wal_store, NodeId(0)).unwrap(); // and ensure voting second time for 1 is not ok vote_resp = sk.process_msg(&vote_request); @@ -1013,12 +1035,11 @@ mod tests { #[test] fn test_epoch_switch() { let storage = InMemoryState { - persisted_state: SafeKeeperState::empty(), + persisted_state: test_sk_state(), }; let wal_store = DummyWalStore { lsn: Lsn(0) }; - let ztli = ZTimelineId::from([0u8; 16]); - let mut sk = SafeKeeper::new(ztli, storage, wal_store, NodeId(0)).unwrap(); + let mut sk = SafeKeeper::new(storage, wal_store, NodeId(0)).unwrap(); let mut ar_hdr = AppendRequestHeader { term: 1, diff --git a/safekeeper/src/send_wal.rs b/safekeeper/src/send_wal.rs index 293cf67c57..db3b931f08 100644 --- a/safekeeper/src/send_wal.rs +++ b/safekeeper/src/send_wal.rs @@ -2,7 +2,7 @@ //! with the "START_REPLICATION" message. use crate::handler::SafekeeperPostgresHandler; -use crate::timeline::{ReplicaState, Timeline, TimelineTools}; +use crate::timeline::{ReplicaState, Timeline}; use crate::wal_storage::WalReader; use anyhow::{bail, Context, Result}; @@ -78,7 +78,13 @@ struct ReplicationConnGuard { impl Drop for ReplicationConnGuard { fn drop(&mut self) { - self.timeline.remove_replica(self.replica); + let res = self.timeline.remove_replica(self.replica); + if let Err(e) = res { + warn!( + "Failed to remove replica {} from timeline {}: {}", + self.replica, self.timeline.zttid, e + ); + } } } @@ -112,7 +118,7 @@ impl ReplicationConn { // Note: deserializing is on m[1..] because we skip the tag byte. state.hs_feedback = HotStandbyFeedback::des(&m[1..]) .context("failed to deserialize HotStandbyFeedback")?; - timeline.update_replica_state(replica_id, state); + timeline.update_replica_state(replica_id, state)?; } Some(STANDBY_STATUS_UPDATE_TAG_BYTE) => { let _reply = StandbyReply::des(&m[1..]) @@ -134,7 +140,7 @@ impl ReplicationConn { // This replica is the source of information to resend to compute. state.pageserver_feedback = Some(reply); - timeline.update_replica_state(replica_id, state); + timeline.update_replica_state(replica_id, state)?; } _ => warn!("unexpected message {:?}", msg), } @@ -167,14 +173,16 @@ impl ReplicationConn { ) -> Result<()> { let _enter = info_span!("WAL sender", timeline = %spg.ztimelineid.unwrap()).entered(); + let tli = Arc::clone(spg.timeline.as_ref().unwrap()); + // spawn the background thread which receives HotStandbyFeedback messages. - let bg_timeline = Arc::clone(spg.timeline.get()); + let bg_timeline = Arc::clone(&tli); let bg_stream_in = self.stream_in.take().unwrap(); let bg_timeline_id = spg.ztimelineid.unwrap(); let state = ReplicaState::new(); // This replica_id is used below to check if it's time to stop replication. - let replica_id = bg_timeline.add_replica(state); + let replica_id = bg_timeline.add_replica(state)?; // Use a guard object to remove our entry from the timeline, when the background // thread and us have both finished using it. @@ -201,11 +209,8 @@ impl ReplicationConn { .build()?; runtime.block_on(async move { - let (inmem_state, persisted_state) = spg.timeline.get().get_state(); + let (inmem_state, persisted_state) = tli.get_state()?; // add persisted_state.timeline_start_lsn == Lsn(0) check - if persisted_state.server.wal_seg_size == 0 { - bail!("Cannot start replication before connecting to walproposer"); - } // Walproposer gets special handling: safekeeper must give proposer all // local WAL till the end, whether committed or not (walproposer will @@ -217,7 +222,7 @@ impl ReplicationConn { // on this safekeeper itself. That's ok as (old) proposer will never be // able to commit such WAL. let stop_pos: Option = if spg.is_walproposer_recovery() { - let wal_end = spg.timeline.get().get_end_of_wal(); + let wal_end = tli.get_flush_lsn()?; Some(wal_end) } else { None @@ -231,7 +236,7 @@ impl ReplicationConn { let mut end_pos = stop_pos.unwrap_or(inmem_state.commit_lsn); let mut wal_reader = WalReader::new( - spg.conf.timeline_dir(&spg.timeline.get().zttid), + spg.conf.timeline_dir(&tli.zttid), &persisted_state, start_pos, spg.conf.wal_backup_enabled, @@ -241,7 +246,7 @@ impl ReplicationConn { let mut send_buf = vec![0u8; MAX_SEND_SIZE]; // watcher for commit_lsn updates - let mut commit_lsn_watch_rx = spg.timeline.get().get_commit_lsn_watch_rx(); + let mut commit_lsn_watch_rx = tli.get_commit_lsn_watch_rx(); loop { if let Some(stop_pos) = stop_pos { @@ -258,7 +263,7 @@ impl ReplicationConn { } else { // TODO: also check once in a while whether we are walsender // to right pageserver. - if spg.timeline.get().stop_walsender(replica_id)? { + if tli.should_walsender_stop(replica_id)? { // Shut down, timeline is suspended. // TODO create proper error type for this bail!("end streaming to {:?}", spg.appname); diff --git a/safekeeper/src/timeline.rs b/safekeeper/src/timeline.rs index 8d101e6ff6..e081cc4f29 100644 --- a/safekeeper/src/timeline.rs +++ b/safekeeper/src/timeline.rs @@ -1,21 +1,23 @@ -//! This module contains timeline id -> safekeeper state map with file-backed -//! persistence and support for interaction between sending and receiving wal. +//! This module implements Timeline lifecycle management and has all neccessary code +//! to glue together SafeKeeper and all other background services. -use anyhow::{bail, Context, Result}; +use anyhow::{anyhow, bail, Context, Result}; use etcd_broker::subscription_value::SkTimelineInfo; -use once_cell::sync::Lazy; -use postgres_ffi::XLogSegNo; +use postgres_ffi::v14::xlog_utils::XLogSegNo; use serde::Serialize; use tokio::sync::watch; use std::cmp::{max, min}; -use std::collections::{HashMap, HashSet}; -use std::fs::{self}; +use std::collections::HashMap; +use std::fs; -use std::sync::{Arc, Mutex, MutexGuard}; +use parking_lot::{MappedMutexGuard, Mutex, MutexGuard}; +use std::io; +use std::path::PathBuf; +use std::time::Instant; use tokio::sync::mpsc::Sender; use tracing::*; @@ -26,12 +28,12 @@ use utils::{ zid::{NodeId, ZTenantId, ZTenantTimelineId}, }; -use crate::control_file; use crate::safekeeper::{ AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState, - SafekeeperMemState, + SafekeeperMemState, ServerInfo, }; use crate::send_wal::HotStandbyFeedback; +use crate::{control_file, GlobalTimelines}; use crate::metrics::FullTimelineInfo; use crate::wal_storage; @@ -95,17 +97,19 @@ struct SharedState { } impl SharedState { - /// Initialize timeline state, creating control file + /// Initialize timeline state, create a control file on disk. fn create( conf: &SafeKeeperConf, zttid: &ZTenantTimelineId, - peer_ids: Vec, + state: SafeKeeperState, ) -> Result { - let state = SafeKeeperState::new(zttid, peer_ids); - let control_store = control_file::FileStorage::create_new(zttid, conf, state)?; + if state.server.wal_seg_size == 0 { + bail!(TimelineError::UninitializedWalSegSize(*zttid)); + } - let wal_store = wal_storage::PhysicalStorage::new(zttid, conf); - let sk = SafeKeeper::new(zttid.timeline_id, control_store, wal_store, conf.my_id)?; + let control_store = control_file::FileStorage::create_new(zttid, conf, state)?; + let wal_store = wal_storage::PhysicalStorage::new(zttid, conf, &control_store)?; + let sk = SafeKeeper::new(control_store, wal_store, conf.my_id)?; Ok(Self { sk, @@ -117,16 +121,25 @@ impl SharedState { }) } - /// Restore SharedState from control file. - /// If file doesn't exist, bails out. + /// Restore SharedState from control file. If file doesn't exist, bails out. fn restore(conf: &SafeKeeperConf, zttid: &ZTenantTimelineId) -> Result { - let control_store = control_file::FileStorage::restore_new(zttid, conf)?; - let wal_store = wal_storage::PhysicalStorage::new(zttid, conf); + let control_store = control_file::FileStorage::restore_new(zttid, conf).map_err(|e| { + if let Some(e) = e.downcast_ref::() { + if e.kind() == io::ErrorKind::NotFound { + return anyhow!(TimelineError::NotFound(*zttid)); + } + } + e + })?; - info!("timeline {} restored", zttid.timeline_id); + if control_store.server.wal_seg_size == 0 { + bail!(TimelineError::UninitializedWalSegSize(*zttid)); + } + + let wal_store = wal_storage::PhysicalStorage::new(zttid, conf, &control_store)?; Ok(Self { - sk: SafeKeeper::new(zttid.timeline_id, control_store, wal_store, conf.my_id)?, + sk: SafeKeeper::new(control_store, wal_store, conf.my_id)?, replicas: Vec::new(), wal_backup_active: false, active: false, @@ -134,6 +147,8 @@ impl SharedState { last_removed_segno: 0, }) } + + fn is_active(&self) -> bool { self.is_wal_backup_required() // FIXME: add tracking of relevant pageservers and check them here individually, @@ -254,7 +269,47 @@ impl SharedState { } } -/// Database instance (tenant) +enum TimelineState { + /// Timeline on-disk state is unknown. We either haven't tried to restore + /// timeline state from disk or got an error. + Uninitialized(UninitializedState), + /// Timeline exists on disk and loaded in memory. + Loaded(Box), + /// Timeline was deleted and cannot be used anymore. + Deleted, +} + +impl TimelineState { + fn inner_mut(&mut self) -> &mut SharedState { + match self { + TimelineState::Loaded(state) => state, + _ => panic!("timeline state is not initialized"), + } + } +} + +#[derive(Debug)] +struct UninitializedState { + /// Safekeeper config. + conf: Box, + /// Error that occurred on last attempt to restore timeline state from disk. + restore_error: Option, + /// Timestamp of the last restore attempt. + last_restore_attempt: Option, +} + +#[derive(Debug, thiserror::Error)] +pub enum TimelineError { + #[error("Timeline {0} was deleted and cannot be used anymore")] + Deleted(ZTenantTimelineId), + #[error("Timeline {0} is not initialized, wal_seg_size is zero")] + UninitializedWalSegSize(ZTenantTimelineId), + #[error("Timeline {0} was not found on disk")] + NotFound(ZTenantTimelineId), +} + +/// Timeline struct manages lifecycle (creation, deletion, restore) of a safekeeper timeline. +/// It also holds SharedState and provides mutually exclusive access to it. pub struct Timeline { pub zttid: ZTenantTimelineId, /// Sending here asks for wal backup launcher attention (start/stop @@ -264,33 +319,170 @@ pub struct Timeline { commit_lsn_watch_tx: watch::Sender, /// For breeding receivers. commit_lsn_watch_rx: watch::Receiver, - mutex: Mutex, + mutex: Mutex, } impl Timeline { - fn new( + /// Create a new uninitialized timeline. Timeline will be tried to restore from disk + /// automatically on most function calls. + pub fn new( + conf: SafeKeeperConf, zttid: ZTenantTimelineId, wal_backup_launcher_tx: Sender, - shared_state: SharedState, ) -> Timeline { - let (commit_lsn_watch_tx, commit_lsn_watch_rx) = - watch::channel(shared_state.sk.inmem.commit_lsn); + let (commit_lsn_watch_tx, commit_lsn_watch_rx) = watch::channel(Lsn::INVALID); Timeline { zttid, wal_backup_launcher_tx, commit_lsn_watch_tx, commit_lsn_watch_rx, - mutex: Mutex::new(shared_state), + mutex: Mutex::new(TimelineState::Uninitialized(UninitializedState { + conf: Box::new(conf), + restore_error: None, + last_restore_attempt: None, + })), } } + /// Try to restore timeline state from disk. + fn load_from_disk( + zttid: &ZTenantTimelineId, + uninit: &mut UninitializedState, + ) -> Result { + info!("Restoring timeline {} from disk", zttid); + + let res = SharedState::restore(&uninit.conf, zttid); + if let Err(e) = &res { + uninit.restore_error = Some(format!("{}", e)); + uninit.last_restore_attempt = Some(Instant::now()); + error!("Failed to restore timeline {} from disk: {}", zttid, e); + } + res + } + + /// Try to create a new timeline on disk. + fn create_on_disk( + zttid: &ZTenantTimelineId, + uninit: &mut UninitializedState, + state: SafeKeeperState, + ) -> Result { + let conf = &uninit.conf; + info!("Creating timeline {} on disk", zttid); + + // TODO: check directory existence + let dir = conf.timeline_dir(zttid); + fs::create_dir_all(dir)?; + + SharedState::create(conf, zttid, state).context("failed to create shared state") + } + + /// Initialize timeline with shared_state. + fn set_shared_state( + &self, + state: SharedState, + state_lock: &mut MutexGuard, + ) -> Result<()> { + assert!(matches!(&**state_lock, TimelineState::Uninitialized(_))); + self.commit_lsn_watch_tx.send(state.sk.inmem.commit_lsn)?; + **state_lock = TimelineState::Loaded(Box::new(state)); + Ok(()) + } + + /// Require timeline state to be loaded. If it's not loaded, try to restore it from disk. + fn require_loaded(&self) -> Result> { + let mut state = self.mutex.lock(); + match &mut *state { + TimelineState::Loaded(_) => {} + TimelineState::Uninitialized(uninit) => { + if let Some(err) = &uninit.restore_error { + // We have an error from last restore attempt, next attempt will not help + // unless something was changed on disk manually. If we will try to restore + // every time, we will have a lot of spam in the logs. + bail!(err.clone()) + } + // TODO: we can allow restoring once a minute, to automatically recover from + // restore failure when something was changed on disk. + + let shared_state = Self::load_from_disk(&self.zttid, uninit)?; + self.set_shared_state(shared_state, &mut state)?; + } + TimelineState::Deleted => bail!(TimelineError::Deleted(self.zttid)), + } + + Ok(MutexGuard::map(state, TimelineState::inner_mut)) + } + + /// Try to load timeline state from disk. If timeline control file is not found, + /// create and initialize a state for the new timeline. + pub fn init_create_or_load(&self, server_info: ServerInfo) -> Result<()> { + let mut state_lock = self.mutex.lock(); + match &mut *state_lock { + TimelineState::Uninitialized(uninit) => { + let shared_state = Self::load_from_disk(&self.zttid, uninit); + match shared_state { + Ok(shared_state) => { + // restored successfully + self.set_shared_state(shared_state, &mut state_lock)?; + Ok(()) + } + Err(e) => { + if let Some(e) = e.downcast_ref::() { + if matches!(e, TimelineError::NotFound(_)) { + // timeline does not exist on disk, create new one + let state = SafeKeeperState::new(&self.zttid, server_info, vec![]); + let shared_state = + Self::create_on_disk(&self.zttid, uninit, state)?; + self.set_shared_state(shared_state, &mut state_lock)?; + return Ok(()); + } + } + Err(e) + } + } + } + TimelineState::Loaded(_) => Ok(()), + TimelineState::Deleted => bail!(TimelineError::Deleted(self.zttid)), + } + } + + /// Deactivates and marks timeline as deleted. Returns whether the timeline was + /// already active. The timeline can no longer be used after this call, almost + /// all functions will return `TimelineError::Deleted`. + /// + /// We assume all threads will stop by themselves eventually (possibly with errors, + /// but no panics). There should be no compute threads (as we're deleting the timeline), + /// actually. Some WAL may be left unsent, but we're deleting the timeline anyway. + async fn delete(&self) -> bool { + let was_active = { + let mut state_lock = self.mutex.lock(); + let was_active = match &*state_lock { + TimelineState::Loaded(state) => state.active, + TimelineState::Uninitialized(_) => false, + + // already deleted + TimelineState::Deleted => return false, + }; + + *state_lock = TimelineState::Deleted; + was_active + }; + + // XXX: we can have a cancellation channel to notify all tasks that they should stop + + let res = self.wal_backup_launcher_tx.send(self.zttid).await; + if let Err(e) = res { + error!("Failed to send stop signal to wal_backup_launcher: {}", e); + } + + was_active + } + /// Register compute connection, starting timeline-related activity if it is /// not running yet. - /// Can fail only if channel to a static thread got closed, which is not normal at all. pub fn on_compute_connect(&self) -> Result<()> { let is_wal_backup_action_pending: bool; { - let mut shared_state = self.mutex.lock().unwrap(); + let mut shared_state = self.require_loaded()?; shared_state.num_computes += 1; is_wal_backup_action_pending = shared_state.update_status(self.zttid); } @@ -303,11 +495,10 @@ impl Timeline { /// De-register compute connection, shutting down timeline activity if /// pageserver doesn't need catchup. - /// Can fail only if channel to a static thread got closed, which is not normal at all. pub fn on_compute_disconnect(&self) -> Result<()> { let is_wal_backup_action_pending: bool; { - let mut shared_state = self.mutex.lock().unwrap(); + let mut shared_state = self.require_loaded()?; shared_state.num_computes -= 1; is_wal_backup_action_pending = shared_state.update_status(self.zttid); } @@ -318,10 +509,10 @@ impl Timeline { Ok(()) } - /// Whether we still need this walsender running? + /// Returns true if walsender should stop sending WAL to pageserver. /// TODO: check this pageserver is actually interested in this timeline. - pub fn stop_walsender(&self, replica_id: usize) -> Result { - let mut shared_state = self.mutex.lock().unwrap(); + pub fn should_walsender_stop(&self, replica_id: usize) -> Result { + let mut shared_state = self.require_loaded()?; if shared_state.num_computes == 0 { let replica_state = shared_state.replicas[replica_id].unwrap(); let stop = shared_state.sk.inmem.commit_lsn == Lsn(0) || // no data at all yet @@ -338,64 +529,54 @@ impl Timeline { /// Returns whether s3 offloading is required and sets current status as /// matching it. pub fn wal_backup_attend(&self) -> bool { - let mut shared_state = self.mutex.lock().unwrap(); + let shared_state = self.require_loaded(); + if shared_state.is_err() { + return false; + } + let mut shared_state = shared_state.unwrap(); shared_state.wal_backup_attend() } - // Can this safekeeper offload to s3? Recently joined safekeepers might not - // have necessary WAL. + /// Can this safekeeper offload to s3? Recently joined safekeepers might not + /// have necessary WAL. pub fn can_wal_backup(&self) -> bool { - self.mutex.lock().unwrap().can_wal_backup() + self.require_loaded() + .map(|state| state.can_wal_backup()) + .unwrap_or(false) } - /// Deactivates the timeline, assuming it is being deleted. - /// Returns whether the timeline was already active. - /// - /// We assume all threads will stop by themselves eventually (possibly with errors, but no panics). - /// There should be no compute threads (as we're deleting the timeline), actually. Some WAL may be left unsent, but - /// we're deleting the timeline anyway. - pub async fn deactivate_for_delete(&self) -> Result { - let was_active: bool; - { - let shared_state = self.mutex.lock().unwrap(); - was_active = shared_state.active; - } - self.wal_backup_launcher_tx.send(self.zttid).await?; - Ok(was_active) - } - - fn is_active(&self) -> bool { - let shared_state = self.mutex.lock().unwrap(); - shared_state.active - } - - /// Returns full timeline info, required for the metrics. - /// If the timeline is not active, returns None instead. + /// Returns full timeline info, required for the metrics. If the timeline is + /// not active, returns None instead. pub fn info_for_metrics(&self) -> Option { - let shared_state = self.mutex.lock().unwrap(); - if !shared_state.active { - return None; - } - - Some(FullTimelineInfo { - zttid: self.zttid, - replicas: shared_state - .replicas - .iter() - .filter_map(|r| r.as_ref()) - .copied() - .collect(), - wal_backup_active: shared_state.wal_backup_active, - timeline_is_active: shared_state.active, - num_computes: shared_state.num_computes, - last_removed_segno: shared_state.last_removed_segno, - epoch_start_lsn: shared_state.sk.epoch_start_lsn, - mem_state: shared_state.sk.inmem.clone(), - persisted_state: shared_state.sk.state.clone(), - flush_lsn: shared_state.sk.wal_store.flush_lsn(), - }) + self.require_loaded() + .map(|state| { + if state.active { + Some(FullTimelineInfo { + zttid: self.zttid, + replicas: state + .replicas + .iter() + .filter_map(|r| r.as_ref()) + .copied() + .collect(), + wal_backup_active: state.wal_backup_active, + timeline_is_active: state.active, + num_computes: state.num_computes, + last_removed_segno: state.last_removed_segno, + epoch_start_lsn: state.sk.epoch_start_lsn, + mem_state: state.sk.inmem.clone(), + persisted_state: state.sk.state.clone(), + flush_lsn: state.sk.wal_store.flush_lsn(), + }) + } else { + None + } + }) + .unwrap_or(None) } + /// Returns commit_lsn watch channel. Channel should be obtained only after + /// the timeline is loaded. Otherwise, the channel will have invalid LSN value. pub fn get_commit_lsn_watch_rx(&self) -> watch::Receiver { self.commit_lsn_watch_rx.clone() } @@ -408,7 +589,7 @@ impl Timeline { let mut rmsg: Option; let commit_lsn: Lsn; { - let mut shared_state = self.mutex.lock().unwrap(); + let mut shared_state = self.require_loaded()?; rmsg = shared_state.sk.process_msg(msg)?; // if this is AppendResponse, fill in proper hot standby feedback and disk consistent lsn @@ -426,29 +607,42 @@ impl Timeline { Ok(rmsg) } - pub fn get_wal_seg_size(&self) -> usize { - self.mutex.lock().unwrap().get_wal_seg_size() + /// Returns wal_seg_size or None if timeline is not loaded. + pub fn get_wal_seg_size(&self) -> Result { + self.require_loaded().map(|state| state.get_wal_seg_size()) } - pub fn get_state(&self) -> (SafekeeperMemState, SafeKeeperState) { - let shared_state = self.mutex.lock().unwrap(); - (shared_state.sk.inmem.clone(), shared_state.sk.state.clone()) + /// Returns true only if the timeline is loaded and active. + pub fn is_active(&self) -> bool { + self.require_loaded() + .map(|state| state.active) + .unwrap_or(false) } - pub fn get_wal_backup_lsn(&self) -> Lsn { - self.mutex.lock().unwrap().sk.inmem.backup_lsn + /// Returns state of the timeline or None if timeline is not loaded. + pub fn get_state(&self) -> Result<(SafekeeperMemState, SafeKeeperState)> { + self.require_loaded() + .map(|state| (state.sk.inmem.clone(), state.sk.state.clone())) } - pub fn set_wal_backup_lsn(&self, backup_lsn: Lsn) { - self.mutex.lock().unwrap().sk.inmem.backup_lsn = backup_lsn; + /// Returns backup_lsn or None if timeline is not loaded. + pub fn get_wal_backup_lsn(&self) -> Result { + self.require_loaded().map(|state| state.sk.inmem.backup_lsn) + } + + /// Sets backup_lsn to the given value. + pub fn set_wal_backup_lsn(&self, backup_lsn: Lsn) -> Result<()> { + let mut shared_state = self.require_loaded()?; + shared_state.sk.inmem.backup_lsn = backup_lsn; // we should check whether to shut down offloader, but this will be done // soon by peer communication anyway. + Ok(()) } - /// Prepare public safekeeper info for reporting. - pub fn get_public_info(&self, conf: &SafeKeeperConf) -> SkTimelineInfo { - let shared_state = self.mutex.lock().unwrap(); - SkTimelineInfo { + /// Return public safekeeper info for broadcasting to broker and other peers. + pub fn get_public_info(&self, conf: &SafeKeeperConf) -> Result { + let shared_state = self.require_loaded()?; + Ok(SkTimelineInfo { last_log_term: Some(shared_state.sk.get_epoch()), flush_lsn: Some(shared_state.sk.wal_store.flush_lsn()), // note: this value is not flushed to control file yet and can be lost @@ -461,7 +655,7 @@ impl Timeline { peer_horizon_lsn: Some(shared_state.sk.inmem.peer_horizon_lsn), safekeeper_connstr: Some(conf.listen_pg_addr.clone()), backup_lsn: Some(shared_state.sk.inmem.backup_lsn), - } + }) } /// Update timeline state with peer safekeeper data. @@ -473,12 +667,7 @@ impl Timeline { let is_wal_backup_action_pending: bool; let commit_lsn: Lsn; { - let mut shared_state = self.mutex.lock().unwrap(); - // WAL seg size not initialized yet (no message from compute ever - // received), can't do much without it. - if shared_state.get_wal_seg_size() == 0 { - return Ok(()); - } + let mut shared_state = self.require_loaded()?; shared_state.sk.record_safekeeper_info(sk_info)?; is_wal_backup_action_pending = shared_state.update_status(self.zttid); commit_lsn = shared_state.sk.inmem.commit_lsn; @@ -491,36 +680,40 @@ impl Timeline { Ok(()) } - pub fn add_replica(&self, state: ReplicaState) -> usize { - let mut shared_state = self.mutex.lock().unwrap(); - shared_state.add_replica(state) + /// Add send_wal replica to the in-memory vector of replicas. + pub fn add_replica(&self, state: ReplicaState) -> Result { + let mut shared_state = self.require_loaded()?; + Ok(shared_state.add_replica(state)) } - pub fn update_replica_state(&self, id: usize, state: ReplicaState) { - let mut shared_state = self.mutex.lock().unwrap(); + /// Update replication replica state. + pub fn update_replica_state(&self, id: usize, state: ReplicaState) -> Result<()> { + let mut shared_state = self.require_loaded()?; shared_state.replicas[id] = Some(state); + Ok(()) } - pub fn remove_replica(&self, id: usize) { - let mut shared_state = self.mutex.lock().unwrap(); + /// Remove send_wal replica from the in-memory vector of replicas. + pub fn remove_replica(&self, id: usize) -> Result<()> { + let mut shared_state = self.require_loaded()?; assert!(shared_state.replicas[id].is_some()); shared_state.replicas[id] = None; + Ok(()) } - pub fn get_end_of_wal(&self) -> Lsn { - let shared_state = self.mutex.lock().unwrap(); - shared_state.sk.wal_store.flush_lsn() + /// Returns flush_lsn. + pub fn get_flush_lsn(&self) -> Result { + let shared_state = self.require_loaded()?; + Ok(shared_state.sk.wal_store.flush_lsn()) } + /// Delete WAL segments from disk that are no longer needed. This is determined + /// based on pageserver's remote_consistent_lsn and local backup_lsn/peer_lsn. pub fn remove_old_wal(&self, wal_backup_enabled: bool) -> Result<()> { let horizon_segno: XLogSegNo; let remover: Box Result<(), anyhow::Error>>; { - let shared_state = self.mutex.lock().unwrap(); - // WAL seg size not initialized yet, no WAL exists. - if shared_state.get_wal_seg_size() == 0 { - return Ok(()); - } + let shared_state = self.require_loaded()?; horizon_segno = shared_state.sk.get_horizon_segno(wal_backup_enabled); remover = shared_state.sk.wal_store.remove_up_to(); if horizon_segno <= 1 || horizon_segno <= shared_state.last_removed_segno { @@ -528,244 +721,105 @@ impl Timeline { } // release the lock before removing } - let _enter = - info_span!("", tenant = %self.zttid.tenant_id, timeline = %self.zttid.timeline_id) - .entered(); + + // delete old WAL files remover(horizon_segno - 1)?; - self.mutex.lock().unwrap().last_removed_segno = horizon_segno; + + // update last_removed_segno + let mut shared_state = self.require_loaded()?; + shared_state.last_removed_segno = horizon_segno; Ok(()) } } -// Utilities needed by various Connection-like objects -pub trait TimelineTools { - fn set(&mut self, conf: &SafeKeeperConf, zttid: ZTenantTimelineId, create: bool) -> Result<()>; - - fn get(&self) -> &Arc; -} - -impl TimelineTools for Option> { - fn set(&mut self, conf: &SafeKeeperConf, zttid: ZTenantTimelineId, create: bool) -> Result<()> { - *self = Some(GlobalTimelines::get(conf, zttid, create)?); - Ok(()) - } - - fn get(&self) -> &Arc { - self.as_ref().unwrap() - } -} - -struct GlobalTimelinesState { - timelines: HashMap>, - wal_backup_launcher_tx: Option>, -} - -static TIMELINES_STATE: Lazy> = Lazy::new(|| { - Mutex::new(GlobalTimelinesState { - timelines: HashMap::new(), - wal_backup_launcher_tx: None, - }) -}); - #[derive(Clone, Copy, Serialize)] pub struct TimelineDeleteForceResult { pub dir_existed: bool, pub was_active: bool, } -/// A zero-sized struct used to manage access to the global timelines map. -pub struct GlobalTimelines; - -impl GlobalTimelines { - pub fn init(wal_backup_launcher_tx: Sender) { - let mut state = TIMELINES_STATE.lock().unwrap(); - assert!(state.wal_backup_launcher_tx.is_none()); - state.wal_backup_launcher_tx = Some(wal_backup_launcher_tx); +/// Deletes directory and it's contents. Returns false if directory does not exist. +fn delete_dir(path: PathBuf) -> Result { + match std::fs::remove_dir_all(path) { + Ok(_) => Ok(true), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(e) => Err(e.into()), } +} - fn create_internal( - mut state: MutexGuard, - conf: &SafeKeeperConf, - zttid: ZTenantTimelineId, - peer_ids: Vec, - ) -> Result> { - match state.timelines.get(&zttid) { - Some(_) => bail!("timeline {} already exists", zttid), - None => { - // TODO: check directory existence - let dir = conf.timeline_dir(&zttid); - fs::create_dir_all(dir)?; +/// Deactivates and sets TimelineState to Deleted, see `Timeline::delete()`, then deletes the +/// corresponding data directory. We assume all timeline threads will eventually check that +/// timeline is Deleted and terminate without panics. +/// +/// Timeline cannot be recreated because we keep all deleted timelines in memory. It can be +/// accidentally created again if safekeeper will restart and compute will connect to it, but this +/// is very unlikely. +pub async fn delete_force( + conf: &SafeKeeperConf, + zttid: &ZTenantTimelineId, +) -> Result { + info!("deleting timeline {}", zttid); + let timeline = GlobalTimelines::get(*zttid); - let shared_state = SharedState::create(conf, &zttid, peer_ids) - .context("failed to create shared state")?; + let was_active = timeline.delete().await; + Ok(TimelineDeleteForceResult { + dir_existed: delete_dir(conf.timeline_dir(zttid))?, + was_active, + }) +} - let new_tli = Arc::new(Timeline::new( - zttid, - state.wal_backup_launcher_tx.as_ref().unwrap().clone(), - shared_state, - )); - state.timelines.insert(zttid, Arc::clone(&new_tli)); - Ok(new_tli) +/// Deactivates and deletes all timelines for the tenant. Returns map of all timelines which +/// the tenant had, `true` if a timeline was active. There may be a race if new timelines are +/// created simultaneously. In that case the function will return error and the caller should +/// try to delete tenant again later. +pub async fn delete_force_all_for_tenant( + conf: &SafeKeeperConf, + tenant_id: &ZTenantId, +) -> Result> { + info!("deleting all timelines for tenant {}", tenant_id); + let to_delete = GlobalTimelines::get_all_for_tenant(*tenant_id); + + let mut err = None; + + let mut deleted = HashMap::new(); + for tli in &to_delete { + let was_active = tli.delete().await; + let res = delete_dir(conf.timeline_dir(&tli.zttid)); + + match res { + Ok(dir_existed) => { + deleted.insert( + tli.zttid, + TimelineDeleteForceResult { + dir_existed, + was_active, + }, + ); + } + Err(e) => { + error!("failed to delete timeline {}: {}", tli.zttid, e); + // Save error to return later. + err = Some(e); } } } - pub fn create( - conf: &SafeKeeperConf, - zttid: ZTenantTimelineId, - peer_ids: Vec, - ) -> Result> { - let state = TIMELINES_STATE.lock().unwrap(); - GlobalTimelines::create_internal(state, conf, zttid, peer_ids) + // There may be inactive timelines, so delete the whole tenant dir as well. + delete_dir(conf.tenant_dir(tenant_id))?; + + let tlis_after_delete = GlobalTimelines::get_all_for_tenant(*tenant_id); + if tlis_after_delete.len() != to_delete.len() { + // Some timelines were created while we were deleting them, returning error + // to the caller, so it can retry later. + bail!( + "failed to delete all timelines for tenant {}: some timelines were created while we were deleting them", + tenant_id + ); } - /// Get a timeline with control file loaded from the global TIMELINES_STATE.timelines map. - /// If control file doesn't exist and create=false, bails out. - pub fn get( - conf: &SafeKeeperConf, - zttid: ZTenantTimelineId, - create: bool, - ) -> Result> { - let _enter = info_span!("", timeline = %zttid.timeline_id).entered(); - - let mut state = TIMELINES_STATE.lock().unwrap(); - - match state.timelines.get(&zttid) { - Some(result) => Ok(Arc::clone(result)), - None => { - let shared_state = SharedState::restore(conf, &zttid); - - let shared_state = match shared_state { - Ok(shared_state) => shared_state, - Err(error) => { - // TODO: always create timeline explicitly - if error - .root_cause() - .to_string() - .contains("No such file or directory") - && create - { - return GlobalTimelines::create_internal(state, conf, zttid, vec![]); - } else { - return Err(error); - } - } - }; - - let new_tli = Arc::new(Timeline::new( - zttid, - state.wal_backup_launcher_tx.as_ref().unwrap().clone(), - shared_state, - )); - state.timelines.insert(zttid, Arc::clone(&new_tli)); - Ok(new_tli) - } - } - } - - /// Get loaded timeline, if it exists. - pub fn get_loaded(zttid: ZTenantTimelineId) -> Option> { - let state = TIMELINES_STATE.lock().unwrap(); - state.timelines.get(&zttid).map(Arc::clone) - } - - /// Get ZTenantTimelineIDs of all active timelines. - pub fn get_active_timelines() -> HashSet { - let state = TIMELINES_STATE.lock().unwrap(); - state - .timelines - .iter() - .filter(|&(_, tli)| tli.is_active()) - .map(|(zttid, _)| *zttid) - .collect() - } - - /// Return FullTimelineInfo for all active timelines. - pub fn active_timelines_metrics() -> Vec { - let state = TIMELINES_STATE.lock().unwrap(); - state - .timelines - .iter() - .filter_map(|(_, tli)| tli.info_for_metrics()) - .collect() - } - - fn delete_force_internal( - conf: &SafeKeeperConf, - zttid: &ZTenantTimelineId, - was_active: bool, - ) -> Result { - match std::fs::remove_dir_all(conf.timeline_dir(zttid)) { - Ok(_) => Ok(TimelineDeleteForceResult { - dir_existed: true, - was_active, - }), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(TimelineDeleteForceResult { - dir_existed: false, - was_active, - }), - Err(e) => Err(e.into()), - } - } - - /// Deactivates and deletes the timeline, see `Timeline::deactivate_for_delete()`, the deletes - /// the corresponding data directory. - /// We assume all timeline threads do not care about `GlobalTimelines` not containing the timeline - /// anymore, and they will eventually terminate without panics. - /// - /// There are multiple ways the timeline may be accidentally "re-created" (so we end up with two - /// `Timeline` objects in memory): - /// a) a compute node connects after this method is called, or - /// b) an HTTP GET request about the timeline is made and it's able to restore the current state, or - /// c) an HTTP POST request for timeline creation is made after the timeline is already deleted. - /// TODO: ensure all of the above never happens. - pub async fn delete_force( - conf: &SafeKeeperConf, - zttid: &ZTenantTimelineId, - ) -> Result { - info!("deleting timeline {}", zttid); - let timeline = TIMELINES_STATE.lock().unwrap().timelines.remove(zttid); - let mut was_active = false; - if let Some(tli) = timeline { - was_active = tli.deactivate_for_delete().await?; - } - GlobalTimelines::delete_force_internal(conf, zttid, was_active) - } - - /// Deactivates and deletes all timelines for the tenant, see `delete()`. - /// Returns map of all timelines which the tenant had, `true` if a timeline was active. - /// There may be a race if new timelines are created simultaneously. - pub async fn delete_force_all_for_tenant( - conf: &SafeKeeperConf, - tenant_id: &ZTenantId, - ) -> Result> { - info!("deleting all timelines for tenant {}", tenant_id); - let mut to_delete = HashMap::new(); - { - // Keep mutex in this scope. - let timelines = &mut TIMELINES_STATE.lock().unwrap().timelines; - for (&zttid, tli) in timelines.iter() { - if zttid.tenant_id == *tenant_id { - to_delete.insert(zttid, tli.clone()); - } - } - // TODO: test that the correct subset of timelines is removed. It's complicated because they are implicitly created currently. - timelines.retain(|zttid, _| !to_delete.contains_key(zttid)); - } - let mut deleted = HashMap::new(); - for (zttid, timeline) in to_delete { - let was_active = timeline.deactivate_for_delete().await?; - deleted.insert( - zttid, - GlobalTimelines::delete_force_internal(conf, &zttid, was_active)?, - ); - } - // There may be inactive timelines, so delete the whole tenant dir as well. - match std::fs::remove_dir_all(conf.tenant_dir(tenant_id)) { - Ok(_) => (), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => (), - e => e?, - }; + if let Some(e) = err { + Err(e) + } else { Ok(deleted) } } diff --git a/safekeeper/src/timelines_global_map.rs b/safekeeper/src/timelines_global_map.rs new file mode 100644 index 0000000000..66e0550e18 --- /dev/null +++ b/safekeeper/src/timelines_global_map.rs @@ -0,0 +1,72 @@ +//! This module contains global (tenant_id, timeline_id) -> Arc mapping. + +use crate::timeline::Timeline; +use crate::SafeKeeperConf; +use once_cell::sync::Lazy; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use tokio::sync::mpsc::Sender; +use utils::zid::{ZTenantId, ZTenantTimelineId}; + +struct GlobalTimelinesState { + timelines: HashMap>, + wal_backup_launcher_tx: Option>, + conf: SafeKeeperConf, +} + +static TIMELINES_STATE: Lazy> = Lazy::new(|| { + Mutex::new(GlobalTimelinesState { + timelines: HashMap::new(), + wal_backup_launcher_tx: None, + conf: SafeKeeperConf::default(), + }) +}); + +/// A zero-sized struct used to manage access to the global timelines map. +pub struct GlobalTimelines; + +impl GlobalTimelines { + // Inject dependencies needed for the timeline constructors. + pub fn init(conf: SafeKeeperConf, wal_backup_launcher_tx: Sender) { + let mut state = TIMELINES_STATE.lock().unwrap(); + assert!(state.wal_backup_launcher_tx.is_none()); + state.wal_backup_launcher_tx = Some(wal_backup_launcher_tx); + state.conf = conf; + } + + /// Get a timeline from the global map. If it doesn't exist in the map, new uninitialised timeline + /// will be added to the map. + pub fn get(zttid: ZTenantTimelineId) -> Arc { + let mut global_lock = TIMELINES_STATE.lock().unwrap(); + + match global_lock.timelines.get(&zttid) { + Some(result) => Arc::clone(result), + None => { + let tli = Arc::new(Timeline::new( + global_lock.conf.clone(), + zttid, + global_lock.wal_backup_launcher_tx.as_ref().unwrap().clone(), + )); + global_lock.timelines.insert(zttid, tli.clone()); + tli + } + } + } + + /// Returns all timelines. This is used for background timeline proccesses. + pub fn get_all() -> Vec> { + let global_lock = TIMELINES_STATE.lock().unwrap(); + global_lock.timelines.values().cloned().collect() + } + + /// Returns all timelines belonging to a given tenant. Used for deleting all timelines of a tenant. + pub fn get_all_for_tenant(tenant_id: ZTenantId) -> Vec> { + let global_lock = TIMELINES_STATE.lock().unwrap(); + global_lock + .timelines + .values() + .filter(|t| t.zttid.tenant_id == tenant_id) + .cloned() + .collect() + } +} diff --git a/safekeeper/src/wal_backup.rs b/safekeeper/src/wal_backup.rs index 5d946e37a4..bb6d0f3f85 100644 --- a/safekeeper/src/wal_backup.rs +++ b/safekeeper/src/wal_backup.rs @@ -26,8 +26,8 @@ use tracing::*; use utils::{lsn::Lsn, zid::ZTenantTimelineId}; use crate::broker::{Election, ElectionLeader}; -use crate::timeline::{GlobalTimelines, Timeline}; -use crate::{broker, SafeKeeperConf}; +use crate::timeline::Timeline; +use crate::{broker, GlobalTimelines, SafeKeeperConf}; use once_cell::sync::OnceCell; @@ -54,7 +54,7 @@ pub fn wal_backup_launcher_thread_main( /// Check whether wal backup is required for timeline. If yes, mark that launcher is /// aware of current status and return the timeline. fn is_wal_backup_required(zttid: ZTenantTimelineId) -> Option> { - GlobalTimelines::get_loaded(zttid).filter(|t| t.wal_backup_attend()) + Some(GlobalTimelines::get(zttid)).filter(|tli| tli.wal_backup_attend()) } struct WalBackupTaskHandle { @@ -191,7 +191,8 @@ struct WalBackupTask { election: Election, } -/// Offload single timeline. +/// Offload single timeline. Called only after we checked that backup +/// is required (wal_backup_attend) and possible (can_wal_backup). async fn backup_task_main( zttid: ZTenantTimelineId, timeline_dir: PathBuf, @@ -199,18 +200,17 @@ async fn backup_task_main( election: Election, ) { info!("started"); - let timeline: Arc = if let Some(tli) = GlobalTimelines::get_loaded(zttid) { - tli - } else { - /* Timeline could get deleted while task was starting, just exit then. */ - info!("no timeline, exiting"); + let tli = GlobalTimelines::get(zttid); + let res = tli.get_wal_seg_size(); + if let Err(e) = res { + info!("backup error for timeline {}: {}", zttid, e); return; - }; + } let mut wb = WalBackupTask { - wal_seg_size: timeline.get_wal_seg_size(), - commit_lsn_watch_rx: timeline.get_commit_lsn_watch_rx(), - timeline, + wal_seg_size: res.unwrap(), + commit_lsn_watch_rx: tli.get_commit_lsn_watch_rx(), + timeline: tli, timeline_dir, leader: None, election, @@ -253,7 +253,7 @@ impl WalBackupTask { if retry_attempt == 0 { // wait for new WAL to arrive if let Err(e) = self.commit_lsn_watch_rx.changed().await { - // should never happen, as we hold Arc to timeline. + // can happen if timeline is deleted error!("commit_lsn watch shut down: {:?}", e); return; } @@ -279,7 +279,12 @@ impl WalBackupTask { continue; /* nothing to do, common case as we wake up on every commit_lsn bump */ } // Perhaps peers advanced the position, check shmem value. - backup_lsn = self.timeline.get_wal_backup_lsn(); + let res = self.timeline.get_wal_backup_lsn(); + if let Err(e) = res { + error!("backup error: {}", e); + return; + } + backup_lsn = res.unwrap(); if backup_lsn.segment_number(self.wal_seg_size) >= commit_lsn.segment_number(self.wal_seg_size) { @@ -322,7 +327,11 @@ impl WalBackupTask { { Ok(backup_lsn_result) => { backup_lsn = backup_lsn_result; - self.timeline.set_wal_backup_lsn(backup_lsn_result); + let res = self.timeline.set_wal_backup_lsn(backup_lsn_result); + if let Err(e) = res { + error!("backup error: {}", e); + return; + } retry_attempt = 0; } Err(e) => { diff --git a/safekeeper/src/wal_storage.rs b/safekeeper/src/wal_storage.rs index 644237a00d..cdf9ecde97 100644 --- a/safekeeper/src/wal_storage.rs +++ b/safekeeper/src/wal_storage.rs @@ -7,7 +7,7 @@ //! //! Note that last file has `.partial` suffix, that's different from postgres. -use anyhow::{anyhow, bail, Context, Result}; +use anyhow::{bail, Context, Result}; use std::io::{self, Seek, SeekFrom}; use std::pin::Pin; use tokio::io::AsyncRead; @@ -17,7 +17,7 @@ use postgres_ffi::v14::xlog_utils::{ find_end_of_wal, IsPartialXLogFileName, IsXLogFileName, XLogFromFileName, }; use postgres_ffi::{XLogSegNo, PG_TLI}; -use std::cmp::min; +use std::cmp::{max, min}; use std::fs::{self, remove_file, File, OpenOptions}; use std::io::Write; @@ -101,9 +101,6 @@ pub trait Storage { /// LSN of last durably stored WAL record. fn flush_lsn(&self) -> Lsn; - /// Init storage with wal_seg_size and read WAL from disk to get latest LSN. - fn init_storage(&mut self, state: &SafeKeeperState) -> Result<()>; - /// Write piece of WAL from buf to disk, but not necessarily sync it. fn write_wal(&mut self, startpos: Lsn, buf: &[u8]) -> Result<()>; @@ -119,7 +116,7 @@ pub trait Storage { } /// PhysicalStorage is a storage that stores WAL on disk. Writes are separated from flushes -/// for better performance. Storage must be initialized before use. +/// for better performance. Storage is initialized in the constructor. /// /// WAL is stored in segments, each segment is a file. Last segment has ".partial" suffix in /// its filename and may be not fully flushed. @@ -127,16 +124,14 @@ pub trait Storage { /// Relationship of LSNs: /// `write_lsn` >= `write_record_lsn` >= `flush_record_lsn` /// -/// When storage is just created, all LSNs are zeroes and there are no segments on disk. +/// When storage is created first time, all LSNs are zeroes and there are no segments on disk. pub struct PhysicalStorage { metrics: WalStorageMetrics, - zttid: ZTenantTimelineId, timeline_dir: PathBuf, conf: SafeKeeperConf, - // fields below are filled upon initialization - /// None if uninitialized, Some(usize) if storage is initialized. - wal_seg_size: Option, + /// Size of WAL segment in bytes. + wal_seg_size: usize, /// Written to disk, but possibly still in the cache and not fully persisted. /// Also can be ahead of record_lsn, if happen to be in the middle of a WAL record. @@ -161,25 +156,49 @@ pub struct PhysicalStorage { } impl PhysicalStorage { - pub fn new(zttid: &ZTenantTimelineId, conf: &SafeKeeperConf) -> PhysicalStorage { + pub fn new( + zttid: &ZTenantTimelineId, + conf: &SafeKeeperConf, + state: &SafeKeeperState, + ) -> Result { let timeline_dir = conf.timeline_dir(zttid); - PhysicalStorage { + + if state.server.wal_seg_size == 0 { + bail!("wal_seg_size must be initialized before creating PhysicalStorage"); + } + let wal_seg_size = state.server.wal_seg_size as usize; + + // Find out where stored WAL ends, starting at commit_lsn which is a + // known recent record boundary (unless we don't have WAL at all). + let write_lsn = if state.commit_lsn == Lsn(0) { + Lsn(0) + } else { + find_end_of_wal(&timeline_dir, wal_seg_size, state.commit_lsn)? + }; + + // TODO: do we really know that write_lsn is fully flushed to disk? + // If not, maybe it's better to call fsync() here to be sure? + let flush_lsn = write_lsn; + + info!( + "initialized storage for timeline {}, flush_lsn={}, commit_lsn={}, peer_horizon_lsn={}", + zttid.timeline_id, flush_lsn, state.commit_lsn, state.peer_horizon_lsn, + ); + if flush_lsn < state.commit_lsn || flush_lsn < state.peer_horizon_lsn { + warn!("timeline {} potential data loss: flush_lsn by find_end_of_wal is less than either commit_lsn or peer_horizon_lsn from control file", zttid.timeline_id); + } + + Ok(PhysicalStorage { metrics: WalStorageMetrics::new(zttid), - zttid: *zttid, timeline_dir, conf: conf.clone(), - wal_seg_size: None, - write_lsn: Lsn(0), - write_record_lsn: Lsn(0), - flush_record_lsn: Lsn(0), - decoder: WalStreamDecoder::new(Lsn(0)), + wal_seg_size, + write_lsn, + write_record_lsn: write_lsn, + flush_record_lsn: flush_lsn, + decoder: WalStreamDecoder::new(write_lsn), file: None, - } - } - - /// Wrapper for flush_lsn updates that also updates metrics. - fn update_flush_lsn(&mut self) { - self.flush_record_lsn = self.write_record_lsn; + }) } /// Call fdatasync if config requires so. @@ -204,9 +223,9 @@ impl PhysicalStorage { /// Open or create WAL segment file. Caller must call seek to the wanted position. /// Returns `file` and `is_partial`. - fn open_or_create(&self, segno: XLogSegNo, wal_seg_size: usize) -> Result<(File, bool)> { + fn open_or_create(&self, segno: XLogSegNo) -> Result<(File, bool)> { let (wal_file_path, wal_file_partial_path) = - wal_file_paths(&self.timeline_dir, segno, wal_seg_size)?; + wal_file_paths(&self.timeline_dir, segno, self.wal_seg_size)?; // Try to open already completed segment if let Ok(file) = OpenOptions::new().write(true).open(&wal_file_path) { @@ -222,24 +241,18 @@ impl PhysicalStorage { .open(&wal_file_partial_path) .with_context(|| format!("Failed to open log file {:?}", &wal_file_path))?; - write_zeroes(&mut file, wal_seg_size)?; + write_zeroes(&mut file, self.wal_seg_size)?; self.fsync_file(&mut file)?; Ok((file, true)) } } /// Write WAL bytes, which are known to be located in a single WAL segment. - fn write_in_segment( - &mut self, - segno: u64, - xlogoff: usize, - buf: &[u8], - wal_seg_size: usize, - ) -> Result<()> { + fn write_in_segment(&mut self, segno: u64, xlogoff: usize, buf: &[u8]) -> Result<()> { let mut file = if let Some(file) = self.file.take() { file } else { - let (mut file, is_partial) = self.open_or_create(segno, wal_seg_size)?; + let (mut file, is_partial) = self.open_or_create(segno)?; assert!(is_partial, "unexpected write into non-partial segment file"); file.seek(SeekFrom::Start(xlogoff as u64))?; file @@ -247,13 +260,13 @@ impl PhysicalStorage { file.write_all(buf)?; - if xlogoff + buf.len() == wal_seg_size { + if xlogoff + buf.len() == self.wal_seg_size { // If we reached the end of a WAL segment, flush and close it. self.fdatasync_file(&mut file)?; // Rename partial file to completed file let (wal_file_path, wal_file_partial_path) = - wal_file_paths(&self.timeline_dir, segno, wal_seg_size)?; + wal_file_paths(&self.timeline_dir, segno, self.wal_seg_size)?; fs::rename(&wal_file_partial_path, &wal_file_path)?; } else { // otherwise, file can be reused later @@ -269,10 +282,6 @@ impl PhysicalStorage { /// /// Updates `write_lsn`. fn write_exact(&mut self, pos: Lsn, mut buf: &[u8]) -> Result<()> { - let wal_seg_size = self - .wal_seg_size - .ok_or_else(|| anyhow!("wal_seg_size is not initialized"))?; - if self.write_lsn != pos { // need to flush the file before discarding it if let Some(mut file) = self.file.take() { @@ -284,17 +293,17 @@ impl PhysicalStorage { while !buf.is_empty() { // Extract WAL location for this block - let xlogoff = self.write_lsn.segment_offset(wal_seg_size) as usize; - let segno = self.write_lsn.segment_number(wal_seg_size); + let xlogoff = self.write_lsn.segment_offset(self.wal_seg_size) as usize; + let segno = self.write_lsn.segment_number(self.wal_seg_size); // If crossing a WAL boundary, only write up until we reach wal segment size. - let bytes_write = if xlogoff + buf.len() > wal_seg_size { - wal_seg_size - xlogoff + let bytes_write = if xlogoff + buf.len() > self.wal_seg_size { + self.wal_seg_size - xlogoff } else { buf.len() }; - self.write_in_segment(segno, xlogoff, &buf[..bytes_write], wal_seg_size)?; + self.write_in_segment(segno, xlogoff, &buf[..bytes_write])?; self.write_lsn += bytes_write as u64; buf = &buf[bytes_write..]; } @@ -309,53 +318,6 @@ impl Storage for PhysicalStorage { self.flush_record_lsn } - /// Storage needs to know wal_seg_size to know which segment to read/write, but - /// wal_seg_size is not always known at the moment of storage creation. This method - /// allows to postpone its initialization. - fn init_storage(&mut self, state: &SafeKeeperState) -> Result<()> { - if state.server.wal_seg_size == 0 { - // wal_seg_size is still unknown. This is dead path normally, should - // be used only in tests. - return Ok(()); - } - - if let Some(wal_seg_size) = self.wal_seg_size { - // physical storage is already initialized - assert_eq!(wal_seg_size, state.server.wal_seg_size as usize); - return Ok(()); - } - - // initialize physical storage - let wal_seg_size = state.server.wal_seg_size as usize; - self.wal_seg_size = Some(wal_seg_size); - - // Find out where stored WAL ends, starting at commit_lsn which is a - // known recent record boundary (unless we don't have WAL at all). - self.write_lsn = if state.commit_lsn == Lsn(0) { - Lsn(0) - } else { - find_end_of_wal(&self.timeline_dir, wal_seg_size, state.commit_lsn)? - }; - - self.write_record_lsn = self.write_lsn; - - // TODO: do we really know that write_lsn is fully flushed to disk? - // If not, maybe it's better to call fsync() here to be sure? - self.update_flush_lsn(); - - info!( - "initialized storage for timeline {}, flush_lsn={}, commit_lsn={}, peer_horizon_lsn={}", - self.zttid.timeline_id, self.flush_record_lsn, state.commit_lsn, state.peer_horizon_lsn, - ); - if self.flush_record_lsn < state.commit_lsn - || self.flush_record_lsn < state.peer_horizon_lsn - { - warn!("timeline {} potential data loss: flush_lsn by find_end_of_wal is less than either commit_lsn or peer_horizon_lsn from control file", self.zttid.timeline_id); - } - - Ok(()) - } - /// Write WAL to disk. fn write_wal(&mut self, startpos: Lsn, buf: &[u8]) -> Result<()> { // Disallow any non-sequential writes, which can result in gaps or overwrites. @@ -419,80 +381,83 @@ impl Storage for PhysicalStorage { // We have unflushed data (write_lsn != flush_lsn), but no file. // This should only happen if last file was fully written and flushed, // but haven't updated flush_lsn yet. - assert!(self.write_lsn.segment_offset(self.wal_seg_size.unwrap()) == 0); + if self.write_lsn.segment_offset(self.wal_seg_size) != 0 { + bail!( + "unexpected unflushed data with no open file, write_lsn={}, flush_lsn={}", + self.write_lsn, + self.flush_record_lsn + ); + } } // everything is flushed now, let's update flush_lsn - self.update_flush_lsn(); + self.flush_record_lsn = self.write_record_lsn; Ok(()) } /// Truncate written WAL by removing all WAL segments after the given LSN. /// end_pos must point to the end of the WAL record. fn truncate_wal(&mut self, end_pos: Lsn) -> Result<()> { - let wal_seg_size = self - .wal_seg_size - .ok_or_else(|| anyhow!("wal_seg_size is not initialized"))?; - // Streaming must not create a hole, so truncate cannot be called on non-written lsn - assert!(self.write_lsn == Lsn(0) || self.write_lsn >= end_pos); + if self.write_lsn != Lsn(0) && end_pos >= self.write_lsn { + bail!( + "truncate_wal called on non-written WAL, write_lsn={}, end_pos={}", + self.write_lsn, + end_pos + ); + } // Close previously opened file, if any if let Some(mut unflushed_file) = self.file.take() { self.fdatasync_file(&mut unflushed_file)?; } - let xlogoff = end_pos.segment_offset(wal_seg_size) as usize; - let segno = end_pos.segment_number(wal_seg_size); - let (mut file, is_partial) = self.open_or_create(segno, wal_seg_size)?; + let xlogoff = end_pos.segment_offset(self.wal_seg_size) as usize; + let segno = end_pos.segment_number(self.wal_seg_size); + + // Remove all segments after the given LSN. + remove_segments_from_disk(&self.timeline_dir, self.wal_seg_size, |x| x > segno)?; + + let (mut file, is_partial) = self.open_or_create(segno)?; // Fill end with zeroes file.seek(SeekFrom::Start(xlogoff as u64))?; - write_zeroes(&mut file, wal_seg_size - xlogoff)?; + write_zeroes(&mut file, self.wal_seg_size - xlogoff)?; self.fdatasync_file(&mut file)?; if !is_partial { // Make segment partial once again let (wal_file_path, wal_file_partial_path) = - wal_file_paths(&self.timeline_dir, segno, wal_seg_size)?; + wal_file_paths(&self.timeline_dir, segno, self.wal_seg_size)?; fs::rename(&wal_file_path, &wal_file_partial_path)?; } - // Remove all subsequent segments - let mut segno = segno; - loop { - segno += 1; - let (wal_file_path, wal_file_partial_path) = - wal_file_paths(&self.timeline_dir, segno, wal_seg_size)?; - // TODO: better use fs::try_exists which is currently available only in nightly build - if wal_file_path.exists() { - fs::remove_file(&wal_file_path)?; - } else if wal_file_partial_path.exists() { - fs::remove_file(&wal_file_partial_path)?; - } else { - break; - } - } - // Update LSNs self.write_lsn = end_pos; self.write_record_lsn = end_pos; - self.update_flush_lsn(); + self.flush_record_lsn = end_pos; Ok(()) } fn remove_up_to(&self) -> Box Result<()>> { let timeline_dir = self.timeline_dir.clone(); - let wal_seg_size = self.wal_seg_size.unwrap(); + let wal_seg_size = self.wal_seg_size; Box::new(move |segno_up_to: XLogSegNo| { - remove_up_to(&timeline_dir, wal_seg_size, segno_up_to) + remove_segments_from_disk(&timeline_dir, wal_seg_size, |x| x <= segno_up_to) }) } } -/// Remove all WAL segments in timeline_dir <= given segno. -fn remove_up_to(timeline_dir: &Path, wal_seg_size: usize, segno_up_to: XLogSegNo) -> Result<()> { +/// Remove all WAL segments in timeline_dir that match the given predicate. +fn remove_segments_from_disk( + timeline_dir: &Path, + wal_seg_size: usize, + remove_predicate: impl Fn(XLogSegNo) -> bool, +) -> Result<()> { let mut n_removed = 0; + let mut min_removed = u64::MAX; + let mut max_removed = u64::MIN; + for entry in fs::read_dir(&timeline_dir)? { let entry = entry?; let entry_path = entry.path(); @@ -504,19 +469,21 @@ fn remove_up_to(timeline_dir: &Path, wal_seg_size: usize, segno_up_to: XLogSegNo continue; } let (segno, _) = XLogFromFileName(fname_str, wal_seg_size); - if segno <= segno_up_to { + if remove_predicate(segno) { remove_file(entry_path)?; n_removed += 1; + min_removed = min(min_removed, segno); + max_removed = max(max_removed, segno); } } } - let segno_from = segno_up_to - n_removed + 1; - info!( - "removed {} WAL segments [{}; {}]", - n_removed, - XLogFileName(PG_TLI, segno_from, wal_seg_size), - XLogFileName(PG_TLI, segno_up_to, wal_seg_size) - ); + + if n_removed > 0 { + info!( + "removed {} WAL segments [{}; {}]", + n_removed, min_removed, max_removed + ); + } Ok(()) } @@ -526,8 +493,10 @@ pub struct WalReader { pos: Lsn, wal_segment: Option>>, - enable_remote_read: bool, // S3 will be used to read WAL if LSN is not available locally + enable_remote_read: bool, + + // We don't have WAL locally if LSN is less than local_start_lsn local_start_lsn: Lsn, }