rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-07 22:20:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor safekeeper.rs

Browse Source
This commit is contained in:
Arthur Petukhovsky
2022-08-02 16:26:16 +00:00
parent e2813b4144
commit f2efe2496e
1 changed files with 47 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
safekeeper/src/safekeeper.rs
View File

@@ -484,8 +484,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<CTRL: control_file::Storage, WAL: wal_storage::Storage> {
/// 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
@@ -508,19 +512,27 @@ where
CTRL: control_file::Storage,
WAL: wal_storage::Storage,
{
// constructor
/// 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(
ztli: ZTimelineId,
state: CTRL,
mut wal_store: WAL,
node_id: NodeId,
) -> Result<SafeKeeper<CTRL, WAL>> {
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);
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,
@@ -579,7 +591,7 @@ where
&mut self,
msg: &ProposerGreeting,
) -> Result<Option<AcceptorProposerMessage>> {
/* Check protocol compatibility */
// Check protocol compatibility
if msg.protocol_version != SK_PROTOCOL_VERSION {
bail!(
"incompatible protocol version {}, expected {}",
@@ -587,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
);
@@ -610,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 {:?}",
@@ -670,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 {
@@ -696,7 +706,12 @@ where
fn handle_elected(&mut self, msg: &ProposerElected) -> Result<Option<AcceptorProposerMessage>> {
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);
@@ -749,7 +764,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);
@@ -952,10 +967,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(())