convert voterequest

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':

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 */

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

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<ProposerAcceptorMessage> {
         // 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)))
     }