From 20e974ecddea86e5f9c1cf270b57b2a4d7f9c058 Mon Sep 17 00:00:00 2001 From: Arseny Sher Date: Wed, 29 Jan 2025 12:31:01 +0100 Subject: [PATCH] convert voterequest --- pgxn/neon/walproposer.c | 32 ++++++++++++++++++++++++-------- pgxn/neon/walproposer.h | 15 +++++++++++++-- safekeeper/src/recovery.rs | 6 +++++- safekeeper/src/safekeeper.rs | 32 ++++++++++++++++++++++++++------ 4 files changed, 68 insertions(+), 17 deletions(-) diff --git a/pgxn/neon/walproposer.c b/pgxn/neon/walproposer.c index bc27f7fa98..1044331be2 100644 --- a/pgxn/neon/walproposer.c +++ b/pgxn/neon/walproposer.c @@ -164,6 +164,7 @@ WalProposerCreate(WalProposerConfig *config, walproposer_api api) void WalProposerFree(WalProposer *wp) { + MembershipConfigurationFree(&wp->mconf); for (int i = 0; i < wp->n_safekeepers; i++) { Safekeeper *sk = &wp->safekeeper[i]; @@ -723,11 +724,9 @@ RecvAcceptorGreeting(Safekeeper *sk) wp->propTerm++; wp_log(LOG, "proposer connected to quorum (%d) safekeepers, propTerm=" INT64_FORMAT, wp->quorum, wp->propTerm); - wp->voteRequest = (VoteRequest) - { - .tag = 'v', - .term = wp->propTerm - }; + wp->voteRequest.pam.tag = 'v'; + wp->voteRequest.generation = wp->mconf.generation; + wp->voteRequest.term = wp->propTerm; } } else if (sk->greetResponse.term > wp->propTerm) @@ -778,7 +777,8 @@ SendVoteRequest(Safekeeper *sk) &sk->outbuf, wp->config->proto_version); /* We have quorum for voting, send our vote request */ - wp_log(LOG, "requesting vote from %s:%s for term " UINT64_FORMAT, sk->host, sk->port, wp->voteRequest.term); + wp_log(LOG, "requesting vote from %s:%s for generation %u term " UINT64_FORMAT, sk->host, sk->port, + wp->voteRequest.generation, wp->voteRequest.term); /* On failure, logging & resetting is handled */ BlockingWrite(sk, sk->outbuf.data, sk->outbuf.len, SS_WAIT_VERDICT); /* If successful, wait for read-ready with SS_WAIT_VERDICT */ @@ -1797,7 +1797,7 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf resetStringInfo(buf); /* removeme tag check after converting all msgs */ - if (proto_version == 3 && msg->tag == 'g') + if (proto_version == 3 && (msg->tag == 'g' || msg->tag == 'v')) { /* * v2 sends structs for some messages as is, so commonly send tag only @@ -1820,6 +1820,15 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf pq_sendint32(buf, m->wal_seg_size); break; } + case 'v': + { + VoteRequest *m = (VoteRequest *) msg; + + pq_sendint32(buf, m->generation); + pq_sendint64(buf, m->term); + break; + + } default: wp_log(FATAL, "unexpected message type %c to serialize", msg->tag); } @@ -1865,7 +1874,14 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf case 'v': { /* v2 sent struct as is */ - pq_sendbytes(buf, (char *) &wp->voteRequest, sizeof(wp->voteRequest)); + VoteRequest *m = (VoteRequest *) msg; + VoteRequestV2 voteRequestV2; + + voteRequestV2.tag = m->pam.tag; + voteRequestV2.term = m->term; + /* removed field */ + memset(&voteRequestV2.proposerId, 0, sizeof(voteRequestV2.proposerId)); + pq_sendbytes(buf, (char *) &voteRequestV2, sizeof(voteRequestV2)); break; } case 'e': diff --git a/pgxn/neon/walproposer.h b/pgxn/neon/walproposer.h index 268f36c627..94897eed06 100644 --- a/pgxn/neon/walproposer.h +++ b/pgxn/neon/walproposer.h @@ -238,11 +238,19 @@ typedef struct AcceptorGreeting * Proposer -> Acceptor vote request. */ typedef struct VoteRequest +{ + ProposerAcceptorMessage pam; /* message tag */ + Generation generation; /* membership conf generation */ + term_t term; +} VoteRequest; + +/* protocol v2 variant, kept while wp supports it */ +typedef struct VoteRequestV2 { uint64 tag; term_t term; pg_uuid_t proposerId; /* for monitoring/debugging */ -} VoteRequest; +} VoteRequestV2; /* Element of term switching chain. */ typedef struct TermSwitchEntry @@ -717,11 +725,14 @@ typedef struct WalProposerConfig typedef struct WalProposer { WalProposerConfig *config; - int n_safekeepers; + /* Current walproposer membership configuration */ + MembershipConfiguration mconf; /* (n_safekeepers / 2) + 1 */ int quorum; + /* Number of occupied slots in safekeepers[] */ + int n_safekeepers; Safekeeper safekeeper[MAX_SAFEKEEPERS]; /* WAL has been generated up to this point */ diff --git a/safekeeper/src/recovery.rs b/safekeeper/src/recovery.rs index 61647c16b0..bf0d30c23a 100644 --- a/safekeeper/src/recovery.rs +++ b/safekeeper/src/recovery.rs @@ -7,6 +7,7 @@ use std::{fmt, pin::pin}; use anyhow::{bail, Context}; use futures::StreamExt; use postgres_protocol::message::backend::ReplicationMessage; +use safekeeper_api::membership::INVALID_GENERATION; use safekeeper_api::models::{PeerInfo, TimelineStatus}; use safekeeper_api::Term; use tokio::sync::mpsc::{channel, Receiver, Sender}; @@ -267,7 +268,10 @@ async fn recover( ); // Now understand our term history. - let vote_request = ProposerAcceptorMessage::VoteRequest(VoteRequest { term: donor.term }); + let vote_request = ProposerAcceptorMessage::VoteRequest(VoteRequest { + generation: INVALID_GENERATION, + term: donor.term, + }); let vote_response = match tli .process_msg(&vote_request) .await diff --git a/safekeeper/src/safekeeper.rs b/safekeeper/src/safekeeper.rs index 8b829a8389..0eaade076b 100644 --- a/safekeeper/src/safekeeper.rs +++ b/safekeeper/src/safekeeper.rs @@ -6,6 +6,7 @@ use bytes::{Buf, BufMut, Bytes, BytesMut}; use postgres_ffi::{TimeLineID, MAX_SEND_SIZE}; use safekeeper_api::membership; +use safekeeper_api::membership::Generation; use safekeeper_api::membership::MemberSet; use safekeeper_api::membership::SafekeeperId; use safekeeper_api::membership::INVALID_GENERATION; @@ -237,8 +238,15 @@ pub struct AcceptorGreeting { } /// Vote request sent from proposer to safekeepers -#[derive(Debug, Deserialize)] +#[derive(Debug)] pub struct VoteRequest { + pub generation: Generation, + pub term: Term, +} + +/// V2 of the message; exists as a struct because we (de)serialized it as is. +#[derive(Debug, Deserialize)] +pub struct VoteRequestV2 { pub term: Term, } @@ -448,7 +456,7 @@ impl ProposerAcceptorMessage { pub fn parse(mut msg_bytes: Bytes, proto_version: u32) -> Result { // TODO remove after converting all msgs let t = msg_bytes[0] as char; - if proto_version == SK_PROTO_VERSION_3 && t == 'g' { + if proto_version == SK_PROTO_VERSION_3 && (t == 'g' || t == 'v') { if msg_bytes.is_empty() { bail!("ProposerAcceptorMessage is not complete: missing tag"); } @@ -481,6 +489,14 @@ impl ProposerAcceptorMessage { }; Ok(ProposerAcceptorMessage::Greeting(g)) } + 'v' => { + let generation = msg_bytes + .get_u32_f() + .with_context(|| "reading generation")?; + let term = msg_bytes.get_u64_f().with_context(|| "reading term")?; + let v = VoteRequest { generation, term }; + Ok(ProposerAcceptorMessage::VoteRequest(v)) + } _ => bail!("unknown proposer-acceptor message tag: {}", tag), } // TODO remove proto_version == 3 after converting all msgs @@ -507,8 +523,12 @@ impl ProposerAcceptorMessage { Ok(ProposerAcceptorMessage::Greeting(g)) } 'v' => { - let msg = VoteRequest::des_from(&mut stream)?; - Ok(ProposerAcceptorMessage::VoteRequest(msg)) + let msg = VoteRequestV2::des_from(&mut stream)?; + let v = VoteRequest { + generation: INVALID_GENERATION, + term: msg.term, + }; + Ok(ProposerAcceptorMessage::VoteRequest(v)) } 'e' => { let mut msg_bytes = stream.into_inner(); @@ -571,7 +591,7 @@ impl ProposerAcceptorMessage { size += match self { Self::Greeting(_) => 0, - Self::VoteRequest(VoteRequest { term: _ }) => 0, + Self::VoteRequest(_) => 0, Self::Elected(ProposerElected { term: _, @@ -904,7 +924,7 @@ where resp.term = self.state.acceptor_state.term; resp.vote_given = true as u64; } - info!("processed VoteRequest for term {}: {:?}", msg.term, &resp); + info!("processed {:?}: sending {:?}", msg, &resp); Ok(Some(AcceptorProposerMessage::VoteResponse(resp))) }