From 2f67637701a91b28dd6543fc7793ab9a329b41c5 Mon Sep 17 00:00:00 2001
From: Vaidas Gasiunas <vaidas.gasiunas@snowflake.com>
Date: Wed, 11 May 2022 19:12:38 +0200
Subject: [PATCH] Fixing the problem with client getting stuck on server
 downgrades:  - Restoring the original check for strict protocol compatibility
 before sending packets  - Resetting the compatibility flag when connection is
 closed, so that the protocol compatibility is checked again for a new
 connection

---
 fdbrpc/FlowTransport.actor.cpp | 16 +++++++---------
 fdbrpc/FlowTransport.h         |  1 -
 2 files changed, 7 insertions(+), 10 deletions(-)

diff --git a/fdbrpc/FlowTransport.actor.cpp b/fdbrpc/FlowTransport.actor.cpp
index ad737d3be4..ef126ef5fd 100644
--- a/fdbrpc/FlowTransport.actor.cpp
+++ b/fdbrpc/FlowTransport.actor.cpp
@@ -816,6 +816,9 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
 				    .errorUnsuppressed(e)
 				    .suppressFor(1.0)
 				    .detail("PeerAddr", self->destination);
+
+				// Since the connection has closed, we need to check the protocol version the next time we connect
+				self->compatible = true;
 			}
 
 			if (self->destination.isPublic() &&
@@ -885,9 +888,9 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
 Peer::Peer(TransportData* transport, NetworkAddress const& destination)
   : transport(transport), destination(destination), compatible(true), outgoingConnectionIdle(true),
     lastConnectTime(0.0), reconnectionDelay(FLOW_KNOBS->INITIAL_RECONNECTION_TIME), peerReferences(-1),
-    incompatibleProtocolVersionNewer(false), bytesReceived(0), bytesSent(0), lastDataPacketSentTime(now()),
-    outstandingReplies(0), pingLatencies(destination.isPublic() ? FLOW_KNOBS->PING_SAMPLE_AMOUNT : 1),
-    lastLoggedTime(0.0), lastLoggedBytesReceived(0), lastLoggedBytesSent(0), timeoutCount(0),
+    bytesReceived(0), bytesSent(0), lastDataPacketSentTime(now()), outstandingReplies(0),
+    pingLatencies(destination.isPublic() ? FLOW_KNOBS->PING_SAMPLE_AMOUNT : 1), lastLoggedTime(0.0),
+    lastLoggedBytesReceived(0), lastLoggedBytesSent(0), timeoutCount(0),
     protocolVersion(Reference<AsyncVar<Optional<ProtocolVersion>>>(new AsyncVar<Optional<ProtocolVersion>>())),
     connectOutgoingCount(0), connectIncomingCount(0), connectFailedCount(0),
     connectLatencies(destination.isPublic() ? FLOW_KNOBS->NETWORK_CONNECT_SAMPLE_AMOUNT : 1) {
@@ -1257,7 +1260,6 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
 	state bool expectConnectPacket = true;
 	state bool compatible = false;
 	state bool incompatiblePeerCounted = false;
-	state bool incompatibleProtocolVersionNewer = false;
 	state NetworkAddress peerAddress;
 	state ProtocolVersion peerProtocolVersion;
 	state Reference<AuthorizedTenants> authorizedTenants = makeReference<AuthorizedTenants>();
@@ -1323,7 +1325,6 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
 						uint64_t connectionId = pkt.connectionId;
 						if (!pkt.protocolVersion.hasObjectSerializerFlag() ||
 						    !pkt.protocolVersion.isCompatible(g_network->protocolVersion())) {
-							incompatibleProtocolVersionNewer = pkt.protocolVersion > g_network->protocolVersion();
 							NetworkAddress addr = pkt.canonicalRemotePort
 							                          ? NetworkAddress(pkt.canonicalRemoteIp(), pkt.canonicalRemotePort)
 							                          : conn->getPeerAddress();
@@ -1383,7 +1384,6 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
 							    .suppressFor(1.0)
 							    .detail("PeerAddr", NetworkAddress(pkt.canonicalRemoteIp(), pkt.canonicalRemotePort));
 							peer->compatible = compatible;
-							peer->incompatibleProtocolVersionNewer = incompatibleProtocolVersionNewer;
 							if (!compatible) {
 								peer->transport->numIncompatibleConnections++;
 								incompatiblePeerCounted = true;
@@ -1401,7 +1401,6 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
 							}
 							peer = transport->getOrOpenPeer(peerAddress, false);
 							peer->compatible = compatible;
-							peer->incompatibleProtocolVersionNewer = incompatibleProtocolVersionNewer;
 							if (!compatible) {
 								peer->transport->numIncompatibleConnections++;
 								incompatiblePeerCounted = true;
@@ -1741,8 +1740,7 @@ static ReliablePacket* sendPacket(TransportData* self,
 
 	// If there isn't an open connection, a public address, or the peer isn't compatible, we can't send
 	if (!peer || (peer->outgoingConnectionIdle && !destination.getPrimaryAddress().isPublic()) ||
-	    (peer->incompatibleProtocolVersionNewer &&
-	     destination.token != Endpoint::wellKnownToken(WLTOKEN_PING_PACKET))) {
+	    (!peer->compatible && destination.token != Endpoint::wellKnownToken(WLTOKEN_PING_PACKET))) {
 		TEST(true); // Can't send to private address without a compatible open connection
 		return nullptr;
 	}
diff --git a/fdbrpc/FlowTransport.h b/fdbrpc/FlowTransport.h
index ceaf3e6f35..b19f3adb88 100644
--- a/fdbrpc/FlowTransport.h
+++ b/fdbrpc/FlowTransport.h
@@ -159,7 +159,6 @@ struct Peer : public ReferenceCounted<Peer> {
 	double lastConnectTime;
 	double reconnectionDelay;
 	int peerReferences;
-	bool incompatibleProtocolVersionNewer;
 	int64_t bytesReceived;
 	int64_t bytesSent;
 	double lastDataPacketSentTime;