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

convert appendresponse

Browse Source
This commit is contained in:
Arseny Sher
2025-02-04 10:11:27 +01:00
parent 9aad617d76
commit 42059b027e
3 changed files with 66 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pgxn/neon/walproposer.c
+25 -8
View File
@@ -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)
pgxn/neon/walproposer.h
+12 -5
View File
@@ -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.
safekeeper/src/safekeeper.rs
+29 -9
View File
@@ -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)));
}