From 42059b027e75be600565c70ebc616c241b6d8437 Mon Sep 17 00:00:00 2001 From: Arseny Sher Date: Tue, 4 Feb 2025 10:02:15 +0100 Subject: [PATCH] convert appendresponse --- pgxn/neon/walproposer.c | 33 +++++++++++++++++++++++-------- pgxn/neon/walproposer.h | 17 +++++++++++----- safekeeper/src/safekeeper.rs | 38 +++++++++++++++++++++++++++--------- 3 files changed, 66 insertions(+), 22 deletions(-) diff --git a/pgxn/neon/walproposer.c b/pgxn/neon/walproposer.c index 6e704ffe49..7ce69ab82b 100644 --- a/pgxn/neon/walproposer.c +++ b/pgxn/neon/walproposer.c @@ -1817,8 +1817,8 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf case 'a': { /* - * Note: this serializes only AppendRequestHeader, caller is - * expected to append WAL data later. + * Note: this serializes only AppendRequestHeader, caller + * is expected to append WAL data later. */ AppendRequestHeader *m = (AppendRequestHeader *) msg; @@ -1896,10 +1896,11 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf pq_sendint64_le(buf, m->termHistory->entries[i].term); pq_sendint64_le(buf, m->termHistory->entries[i].lsn); } - pq_sendint64_le(buf, 0); /* removed timeline_start_lsn */ + pq_sendint64_le(buf, 0); /* removed timeline_start_lsn */ break; } case 'a': + /* * Note: this serializes only AppendRequestHeader, caller is * expected to append WAL data later. @@ -1911,7 +1912,7 @@ PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf appendRequestHeaderV2.tag = m->apm.tag; appendRequestHeaderV2.term = m->term; - appendRequestHeaderV2.epochStartLsn = 0; /* removed field */ + appendRequestHeaderV2.epochStartLsn = 0; /* removed field */ appendRequestHeaderV2.beginLsn = m->beginLsn; appendRequestHeaderV2.endLsn = m->endLsn; appendRequestHeaderV2.commitLsn = m->commitLsn; @@ -2022,8 +2023,7 @@ AsyncReadMessage(Safekeeper *sk, AcceptorProposerMessage *anymsg) s.maxlen = buf_size; s.cursor = 0; - /* removeme tag check after converting all msgs */ - if (wp->config->proto_version == 3 && (anymsg->tag == 'g' || anymsg->tag == 'v')) + if (wp->config->proto_version == 3) { tag = pq_getmsgbyte(&s); if (tag != anymsg->tag) @@ -2064,6 +2064,24 @@ AsyncReadMessage(Safekeeper *sk, AcceptorProposerMessage *anymsg) pq_getmsgend(&s); return true; } + case 'a': + { + AppendResponse *msg = (AppendResponse *) anymsg; + + msg->generation = pq_getmsgint32(&s); + msg->term = pq_getmsgint64(&s); + msg->flushLsn = pq_getmsgint64(&s); + msg->commitLsn = pq_getmsgint64(&s); + msg->hs.ts = pq_getmsgint64(&s); + msg->hs.xmin.value = pq_getmsgint64(&s); + msg->hs.catalog_xmin.value = pq_getmsgint64(&s); + if (s.len > s.cursor) + ParsePageserverFeedbackMessage(wp, &s, &msg->ps_feedback); + else + msg->ps_feedback.present = false; + pq_getmsgend(&s); + return true; + } default: { wp_log(FATAL, "unexpected message tag %c to read", (char) tag); @@ -2071,8 +2089,7 @@ AsyncReadMessage(Safekeeper *sk, AcceptorProposerMessage *anymsg) } } } - /* removeme tag check after converting all msgs */ - else if (wp->config->proto_version == 2 || wp->config->proto_version == 3) + else if (wp->config->proto_version == 2) { tag = pq_getmsgint64_le(&s); if (tag != anymsg->tag) diff --git a/pgxn/neon/walproposer.h b/pgxn/neon/walproposer.h index 6dac23a4f3..eee55f924f 100644 --- a/pgxn/neon/walproposer.h +++ b/pgxn/neon/walproposer.h @@ -271,9 +271,9 @@ typedef struct VoteResponse AcceptorProposerMessage apm; /* - * Membership conf generation. It's redundant because on mismatch - * safekeeper is expected to ERROR the connection, but let's sanity check - * it. + * Membership conf generation. It's not strictly required because on + * mismatch safekeeper is expected to ERROR the connection, but let's + * sanity check it. */ Generation generation; term_t term; @@ -296,7 +296,7 @@ typedef struct VoteResponse typedef struct ProposerElected { AcceptorProposerMessage apm; - Generation generation; /* membership conf generation */ + Generation generation; /* membership conf generation */ term_t term; /* proposer will send since this point */ XLogRecPtr startStreamingAt; @@ -310,7 +310,7 @@ typedef struct ProposerElected typedef struct AppendRequestHeader { AcceptorProposerMessage apm; - Generation generation; /* membership conf generation */ + Generation generation; /* membership conf generation */ term_t term; /* term of the proposer */ XLogRecPtr beginLsn; /* start position of message in WAL */ XLogRecPtr endLsn; /* end position of message in WAL */ @@ -399,6 +399,13 @@ typedef struct AppendResponse { AcceptorProposerMessage apm; + /* + * Membership conf generation. It's not strictly required because on + * mismatch safekeeper is expected to ERROR the connection, but let's + * sanity check it. + */ + Generation generation; + /* * Current term of the safekeeper; if it is higher than proposer's, the * compute is out of date. diff --git a/safekeeper/src/safekeeper.rs b/safekeeper/src/safekeeper.rs index dc75b45e07..cb1b832f79 100644 --- a/safekeeper/src/safekeeper.rs +++ b/safekeeper/src/safekeeper.rs @@ -340,6 +340,9 @@ pub struct AppendRequestHeaderV2 { /// Report safekeeper state to proposer #[derive(Debug, Serialize, Clone)] pub struct AppendResponse { + // Membership conf generation. Not strictly required because on mismatch + // connection is reset, but let's sanity check it. + generation: Generation, // Current term of the safekeeper; if it is higher than proposer's, the // compute is out of date. pub term: Term, @@ -356,8 +359,9 @@ pub struct AppendResponse { } impl AppendResponse { - fn term_only(term: Term) -> AppendResponse { + fn term_only(generation: Generation, term: Term) -> AppendResponse { AppendResponse { + generation, term, flush_lsn: Lsn(0), commit_lsn: Lsn(0), @@ -778,11 +782,7 @@ impl AcceptorProposerMessage { /// Serialize acceptor -> proposer message. pub fn serialize(&self, buf: &mut BytesMut, proto_version: u32) -> Result<()> { - // TODO remove after converting all msgs - if proto_version == SK_PROTO_VERSION_3 - && (matches!(self, AcceptorProposerMessage::Greeting(_)) - || matches!(self, AcceptorProposerMessage::VoteResponse(_))) - { + if proto_version == SK_PROTO_VERSION_3 { match self { AcceptorProposerMessage::Greeting(msg) => { buf.put_u8('g' as u8); @@ -803,11 +803,26 @@ impl AcceptorProposerMessage { buf.put_u64(e.lsn.into()); } } - _ => bail!("not impl"), + AcceptorProposerMessage::AppendResponse(msg) => { + buf.put_u8('a' as u8); + buf.put_u32(msg.generation); + buf.put_u64(msg.term); + buf.put_u64(msg.flush_lsn.into()); + buf.put_u64(msg.commit_lsn.into()); + buf.put_i64(msg.hs_feedback.ts); + buf.put_u64(msg.hs_feedback.xmin); + buf.put_u64(msg.hs_feedback.catalog_xmin); + + // AsyncReadMessage in walproposer.c will not try to decode pageserver_feedback + // if it is not present. + if let Some(ref msg) = msg.pageserver_feedback { + msg.serialize(buf); + } + } } Ok(()) // TODO remove 3 after converting all msgs - } else if proto_version == SK_PROTO_VERSION_2 || proto_version == SK_PROTO_VERSION_3 { + } else if proto_version == SK_PROTO_VERSION_2 { match self { AcceptorProposerMessage::Greeting(msg) => { buf.put_u64_le('g' as u64); @@ -831,6 +846,7 @@ impl AcceptorProposerMessage { buf.put_u64_le(0); } AcceptorProposerMessage::AppendResponse(msg) => { + // v2 didn't have generation buf.put_u64_le('a' as u64); buf.put_u64_le(msg.term); buf.put_u64_le(msg.flush_lsn.into()); @@ -1053,6 +1069,7 @@ where /// Form AppendResponse from current state. fn append_response(&self) -> AppendResponse { let ar = AppendResponse { + generation: self.state.mconf.generation, term: self.state.acceptor_state.term, flush_lsn: self.flush_lsn(), commit_lsn: self.state.commit_lsn, @@ -1246,7 +1263,10 @@ where // If our term is higher, immediately refuse the message. if self.state.acceptor_state.term > msg.h.term { - let resp = AppendResponse::term_only(self.state.acceptor_state.term); + let resp = AppendResponse::term_only( + self.state.mconf.generation, + self.state.acceptor_state.term, + ); return Ok(Some(AcceptorProposerMessage::AppendResponse(resp))); }