fix

pgxn/neon/walproposer.c

@@ -173,10 +173,8 @@ WalProposerCreate(WalProposerConfig *config, walproposer_api api)
 	}
 	wp->quorum = wp->n_safekeepers / 2 + 1;
 
-	if (wp->config->proto_version != 2 && wp->config->proto_version != 3)
+	if (wp->config->proto_version != 3)
 		wp_log(FATAL, "unsupported safekeeper protocol version %d", wp->config->proto_version);
-	if (wp->safekeepers_generation > INVALID_GENERATION && wp->config->proto_version < 3)
-		wp_log(FATAL, "enabling generations requires protocol version 3");
 	wp_log(LOG, "using safekeeper protocol version %d", wp->config->proto_version);
 
 	/* Fill the greeting package */
@@ -2179,183 +2177,79 @@ MembershipConfigurationSerialize(MembershipConfiguration *mconf, StringInfo buf)
 	}
 }
 
-/* Serialize proposer -> acceptor message into buf using specified version */
+/* Serialize proposer -> acceptor message into buf */
 static void
 PAMessageSerialize(WalProposer *wp, ProposerAcceptorMessage *msg, StringInfo buf, int proto_version)
 {
-	/* both version are supported currently until we fully migrate to 3 */
-	Assert(proto_version == 3 || proto_version == 2);
+	/* only version 3 is supported */
+	Assert(proto_version == 3);
 
 	resetStringInfo(buf);
 
-	if (proto_version == 3)
+	/*
+	 * v2 sends structs for some messages as is, so commonly send tag only
+	 * for v3
+	 */
+	pq_sendint8(buf, msg->tag);
+
+	switch (msg->tag)
 	{
-		/*
-		 * v2 sends structs for some messages as is, so commonly send tag only
-		 * for v3
-		 */
-		pq_sendint8(buf, msg->tag);
+		case 'g':
+			{
+				ProposerGreeting *m = (ProposerGreeting *) msg;
 
-		switch (msg->tag)
-		{
-			case 'g':
+				pq_send_ascii_string(buf, m->tenant_id);
+				pq_send_ascii_string(buf, m->timeline_id);
+				MembershipConfigurationSerialize(&m->mconf, buf);
+				pq_sendint32(buf, m->pg_version);
+				pq_sendint64(buf, m->system_id);
+				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;
+
+			}
+		case 'e':
+			{
+				ProposerElected *m = (ProposerElected *) msg;
+
+				pq_sendint32(buf, m->generation);
+				pq_sendint64(buf, m->term);
+				pq_sendint64(buf, m->startStreamingAt);
+				pq_sendint32(buf, m->termHistory->n_entries);
+				for (uint32 i = 0; i < m->termHistory->n_entries; i++)
 				{
-					ProposerGreeting *m = (ProposerGreeting *) msg;
-
-					pq_send_ascii_string(buf, m->tenant_id);
-					pq_send_ascii_string(buf, m->timeline_id);
-					MembershipConfigurationSerialize(&m->mconf, buf);
-					pq_sendint32(buf, m->pg_version);
-					pq_sendint64(buf, m->system_id);
-					pq_sendint32(buf, m->wal_seg_size);
-					break;
+					pq_sendint64(buf, m->termHistory->entries[i].term);
+					pq_sendint64(buf, m->termHistory->entries[i].lsn);
 				}
-			case 'v':
-				{
-					VoteRequest *m = (VoteRequest *) msg;
-
-					pq_sendint32(buf, m->generation);
-					pq_sendint64(buf, m->term);
-					break;
-
-				}
-			case 'e':
-				{
-					ProposerElected *m = (ProposerElected *) msg;
-
-					pq_sendint32(buf, m->generation);
-					pq_sendint64(buf, m->term);
-					pq_sendint64(buf, m->startStreamingAt);
-					pq_sendint32(buf, m->termHistory->n_entries);
-					for (uint32 i = 0; i < m->termHistory->n_entries; i++)
-					{
-						pq_sendint64(buf, m->termHistory->entries[i].term);
-						pq_sendint64(buf, m->termHistory->entries[i].lsn);
-					}
-					break;
-				}
-			case 'a':
-				{
-					/*
-					 * Note: this serializes only AppendRequestHeader, caller
-					 * is expected to append WAL data later.
-					 */
-					AppendRequestHeader *m = (AppendRequestHeader *) msg;
-
-					pq_sendint32(buf, m->generation);
-					pq_sendint64(buf, m->term);
-					pq_sendint64(buf, m->beginLsn);
-					pq_sendint64(buf, m->endLsn);
-					pq_sendint64(buf, m->commitLsn);
-					pq_sendint64(buf, m->truncateLsn);
-					break;
-				}
-			default:
-				wp_log(FATAL, "unexpected message type %c to serialize", msg->tag);
-		}
-		return;
-	}
-
-	if (proto_version == 2)
-	{
-		switch (msg->tag)
-		{
-			case 'g':
-				{
-					/* v2 sent struct as is */
-					ProposerGreeting *m = (ProposerGreeting *) msg;
-					ProposerGreetingV2 greetRequestV2;
-
-					/* Fill also v2 struct. */
-					greetRequestV2.tag = 'g';
-					greetRequestV2.protocolVersion = proto_version;
-					greetRequestV2.pgVersion = m->pg_version;
-
-					/*
-					 * v3 removed this field because it's easier to pass as
-					 * libq or START_WAL_PUSH options
-					 */
-					memset(&greetRequestV2.proposerId, 0, sizeof(greetRequestV2.proposerId));
-					greetRequestV2.systemId = wp->config->systemId;
-					if (*m->timeline_id != '\0' &&
-						!HexDecodeString(greetRequestV2.timeline_id, m->timeline_id, 16))
-						wp_log(FATAL, "could not parse neon.timeline_id, %s", m->timeline_id);
-					if (*m->tenant_id != '\0' &&
-						!HexDecodeString(greetRequestV2.tenant_id, m->tenant_id, 16))
-						wp_log(FATAL, "could not parse neon.tenant_id, %s", m->tenant_id);
-
-					greetRequestV2.timeline = wp->config->pgTimeline;
-					greetRequestV2.walSegSize = wp->config->wal_segment_size;
-
-					pq_sendbytes(buf, (char *) &greetRequestV2, sizeof(greetRequestV2));
-					break;
-				}
-			case 'v':
-				{
-					/* v2 sent struct as is */
-					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':
-				{
-					ProposerElected *m = (ProposerElected *) msg;
-
-					pq_sendint64_le(buf, m->apm.tag);
-					pq_sendint64_le(buf, m->term);
-					pq_sendint64_le(buf, m->startStreamingAt);
-					pq_sendint32_le(buf, m->termHistory->n_entries);
-					for (int i = 0; i < m->termHistory->n_entries; i++)
-					{
-						pq_sendint64_le(buf, m->termHistory->entries[i].term);
-						pq_sendint64_le(buf, m->termHistory->entries[i].lsn);
-					}
-
-					/*
-					 * Removed timeline_start_lsn. Still send it as a valid
-					 * value until safekeepers taking it from term history are
-					 * deployed.
-					 */
-					pq_sendint64_le(buf, m->termHistory->entries[0].lsn);
-					break;
-				}
-			case 'a':
-
+				break;
+			}
+		case 'a':
+			{
 				/*
-				 * Note: this serializes only AppendRequestHeader, caller is
-				 * expected to append WAL data later.
-				 */
-				{
-					/* v2 sent struct as is */
-					AppendRequestHeader *m = (AppendRequestHeader *) msg;
-					AppendRequestHeaderV2 appendRequestHeaderV2;
+					* Note: this serializes only AppendRequestHeader, caller
+					* is expected to append WAL data later.
+					*/
+				AppendRequestHeader *m = (AppendRequestHeader *) msg;
 
-					appendRequestHeaderV2.tag = m->apm.tag;
-					appendRequestHeaderV2.term = m->term;
-					appendRequestHeaderV2.epochStartLsn = 0;	/* removed field */
-					appendRequestHeaderV2.beginLsn = m->beginLsn;
-					appendRequestHeaderV2.endLsn = m->endLsn;
-					appendRequestHeaderV2.commitLsn = m->commitLsn;
-					appendRequestHeaderV2.truncateLsn = m->truncateLsn;
-					/* removed field */
-					memset(&appendRequestHeaderV2.proposerId, 0, sizeof(appendRequestHeaderV2.proposerId));
-
-					pq_sendbytes(buf, (char *) &appendRequestHeaderV2, sizeof(appendRequestHeaderV2));
-					break;
-				}
-
-			default:
-				wp_log(FATAL, "unexpected message type %c to serialize", msg->tag);
-		}
-		return;
+				pq_sendint32(buf, m->generation);
+				pq_sendint64(buf, m->term);
+				pq_sendint64(buf, m->beginLsn);
+				pq_sendint64(buf, m->endLsn);
+				pq_sendint64(buf, m->commitLsn);
+				pq_sendint64(buf, m->truncateLsn);
+				break;
+			}
+		default:
+			wp_log(FATAL, "unexpected message type %c to serialize", msg->tag);
 	}
-	wp_log(FATAL, "unexpected proto_version %d", proto_version);
+	return;
 }
 
 /*
@@ -2449,141 +2343,72 @@ AsyncReadMessage(Safekeeper *sk, AcceptorProposerMessage *anymsg)
 	s.maxlen = buf_size;
 	s.cursor = 0;
 
-	if (wp->config->proto_version == 3)
+	/* only version 3 is supported */
+	Assert(wp->config->proto_version == 3);
+
+	tag = pq_getmsgbyte(&s);
+	if (tag != anymsg->tag)
 	{
-		tag = pq_getmsgbyte(&s);
-		if (tag != anymsg->tag)
-		{
-			wp_log(WARNING, "unexpected message tag %c from node %s:%s in state %s", (char) tag, sk->host,
-				   sk->port, FormatSafekeeperState(sk));
-			ResetConnection(sk);
-			return false;
-		}
-		switch (tag)
-		{
-			case 'g':
-				{
-					AcceptorGreeting *msg = (AcceptorGreeting *) anymsg;
-
-					msg->nodeId = pq_getmsgint64(&s);
-					MembershipConfigurationDeserialize(&msg->mconf, &s);
-					msg->term = pq_getmsgint64(&s);
-					pq_getmsgend(&s);
-					return true;
-				}
-			case 'v':
-				{
-					VoteResponse *msg = (VoteResponse *) anymsg;
-
-					msg->generation = pq_getmsgint32(&s);
-					msg->term = pq_getmsgint64(&s);
-					msg->voteGiven = pq_getmsgbyte(&s);
-					msg->flushLsn = pq_getmsgint64(&s);
-					msg->truncateLsn = pq_getmsgint64(&s);
-					msg->termHistory.n_entries = pq_getmsgint32(&s);
-					msg->termHistory.entries = palloc(sizeof(TermSwitchEntry) * msg->termHistory.n_entries);
-					for (uint32 i = 0; i < msg->termHistory.n_entries; i++)
-					{
-						msg->termHistory.entries[i].term = pq_getmsgint64(&s);
-						msg->termHistory.entries[i].lsn = pq_getmsgint64(&s);
-					}
-					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);
-					return false;
-				}
-		}
+		wp_log(WARNING, "unexpected message tag %c from node %s:%s in state %s", (char) tag, sk->host,
+				sk->port, FormatSafekeeperState(sk));
+		ResetConnection(sk);
+		return false;
 	}
-	else if (wp->config->proto_version == 2)
+	switch (tag)
 	{
-		tag = pq_getmsgint64_le(&s);
-		if (tag != anymsg->tag)
-		{
-			wp_log(WARNING, "unexpected message tag %c from node %s:%s in state %s", (char) tag, sk->host,
-				   sk->port, FormatSafekeeperState(sk));
-			ResetConnection(sk);
-			return false;
-		}
-		switch (tag)
-		{
-			case 'g':
+		case 'g':
+			{
+				AcceptorGreeting *msg = (AcceptorGreeting *) anymsg;
+
+				msg->nodeId = pq_getmsgint64(&s);
+				MembershipConfigurationDeserialize(&msg->mconf, &s);
+				msg->term = pq_getmsgint64(&s);
+				pq_getmsgend(&s);
+				return true;
+			}
+		case 'v':
+			{
+				VoteResponse *msg = (VoteResponse *) anymsg;
+
+				msg->generation = pq_getmsgint32(&s);
+				msg->term = pq_getmsgint64(&s);
+				msg->voteGiven = pq_getmsgbyte(&s);
+				msg->flushLsn = pq_getmsgint64(&s);
+				msg->truncateLsn = pq_getmsgint64(&s);
+				msg->termHistory.n_entries = pq_getmsgint32(&s);
+				msg->termHistory.entries = palloc(sizeof(TermSwitchEntry) * msg->termHistory.n_entries);
+				for (uint32 i = 0; i < msg->termHistory.n_entries; i++)
 				{
-					AcceptorGreeting *msg = (AcceptorGreeting *) anymsg;
-
-					msg->term = pq_getmsgint64_le(&s);
-					msg->nodeId = pq_getmsgint64_le(&s);
-					pq_getmsgend(&s);
-					return true;
+					msg->termHistory.entries[i].term = pq_getmsgint64(&s);
+					msg->termHistory.entries[i].lsn = pq_getmsgint64(&s);
 				}
+				pq_getmsgend(&s);
+				return true;
+			}
+		case 'a':
+			{
+				AppendResponse *msg = (AppendResponse *) anymsg;
 
-			case 'v':
-				{
-					VoteResponse *msg = (VoteResponse *) anymsg;
-
-					msg->term = pq_getmsgint64_le(&s);
-					msg->voteGiven = pq_getmsgint64_le(&s);
-					msg->flushLsn = pq_getmsgint64_le(&s);
-					msg->truncateLsn = pq_getmsgint64_le(&s);
-					msg->termHistory.n_entries = pq_getmsgint32_le(&s);
-					msg->termHistory.entries = palloc(sizeof(TermSwitchEntry) * msg->termHistory.n_entries);
-					for (int i = 0; i < msg->termHistory.n_entries; i++)
-					{
-						msg->termHistory.entries[i].term = pq_getmsgint64_le(&s);
-						msg->termHistory.entries[i].lsn = pq_getmsgint64_le(&s);
-					}
-					pq_getmsgint64_le(&s);	/* timelineStartLsn */
-					pq_getmsgend(&s);
-					return true;
-				}
-
-			case 'a':
-				{
-					AppendResponse *msg = (AppendResponse *) anymsg;
-
-					msg->term = pq_getmsgint64_le(&s);
-					msg->flushLsn = pq_getmsgint64_le(&s);
-					msg->commitLsn = pq_getmsgint64_le(&s);
-					msg->hs.ts = pq_getmsgint64_le(&s);
-					msg->hs.xmin.value = pq_getmsgint64_le(&s);
-					msg->hs.catalog_xmin.value = pq_getmsgint64_le(&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);
-					return false;
-				}
-		}
+				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);
+				return false;
+			}
 	}
-	wp_log(FATAL, "unsupported proto_version %d", wp->config->proto_version);
-	return false;				/* keep the compiler quiet */
 }
 
 /*

test_runner/regress/test_normal_work.py

@@ -10,14 +10,9 @@ if TYPE_CHECKING:
     from fixtures.pageserver.http import PageserverHttpClient
 
 
-def check_tenant(
-    env: NeonEnv, pageserver_http: PageserverHttpClient, safekeeper_proto_version: int
-):
+def check_tenant(env: NeonEnv, pageserver_http: PageserverHttpClient):
     tenant_id, timeline_id = env.create_tenant()
-    config_lines = [
-        f"neon.safekeeper_proto_version = {safekeeper_proto_version}",
-    ]
-    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id, config_lines=config_lines)
+    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
     # we rely upon autocommit after each statement
     res_1 = endpoint.safe_psql_many(
         queries=[
@@ -42,13 +37,10 @@ def check_tenant(
 
 
 @pytest.mark.parametrize("num_timelines,num_safekeepers", [(3, 1)])
-# Test both proto versions until we fully migrate.
-@pytest.mark.parametrize("safekeeper_proto_version", [2, 3])
 def test_normal_work(
     neon_env_builder: NeonEnvBuilder,
     num_timelines: int,
     num_safekeepers: int,
-    safekeeper_proto_version: int,
 ):
     """
     Basic test:
@@ -68,4 +60,4 @@ def test_normal_work(
     pageserver_http = env.pageserver.http_client()
 
     for _ in range(num_timelines):
-        check_tenant(env, pageserver_http, safekeeper_proto_version)
+        check_tenant(env, pageserver_http)

test_runner/regress/test_wal_acceptor_async.py

@@ -540,16 +540,13 @@ def test_recovery_uncommitted(neon_env_builder: NeonEnvBuilder):
     asyncio.run(run_recovery_uncommitted(env))
 
 
-async def run_wal_truncation(env: NeonEnv, safekeeper_proto_version: int):
+async def run_wal_truncation(env: NeonEnv):
     tenant_id = env.initial_tenant
     timeline_id = env.initial_timeline
 
     (sk1, sk2, sk3) = env.safekeepers
 
-    config_lines = [
-        f"neon.safekeeper_proto_version = {safekeeper_proto_version}",
-    ]
-    ep = env.endpoints.create_start("main", config_lines=config_lines)
+    ep = env.endpoints.create_start("main")
     ep.safe_psql("create table t (key int, value text)")
     ep.safe_psql("insert into t select generate_series(1, 100), 'payload'")
 
@@ -568,7 +565,6 @@ async def run_wal_truncation(env: NeonEnv, safekeeper_proto_version: int):
     sk2.start()
     ep = env.endpoints.create_start(
         "main",
-        config_lines=config_lines,
     )
     ep.safe_psql("insert into t select generate_series(1, 200), 'payload'")
 
@@ -587,13 +583,11 @@ async def run_wal_truncation(env: NeonEnv, safekeeper_proto_version: int):
 
 # Simple deterministic test creating tail of WAL on safekeeper which is
 # truncated when majority without this sk elects walproposer starting earlier.
-# Test both proto versions until we fully migrate.
-@pytest.mark.parametrize("safekeeper_proto_version", [2, 3])
-def test_wal_truncation(neon_env_builder: NeonEnvBuilder, safekeeper_proto_version: int):
+def test_wal_truncation(neon_env_builder: NeonEnvBuilder):
     neon_env_builder.num_safekeepers = 3
     env = neon_env_builder.init_start()
 
-    asyncio.run(run_wal_truncation(env, safekeeper_proto_version))
+    asyncio.run(run_wal_truncation(env))
 
 
 async def quorum_sanity_single(