rpc: reset transport version on connect

This is needed for the case in which a peer on the other end of a
reconnect_transport is downgraded. unless we reset the version on
reconnect then we'll get a policy violation if the peer responds
with a lower version than the transport had been upgraded to.

Signed-off-by: Noah Watkins <noahwatkins@gmail.com>

src/v/rpc/test/rpc_gen_cycling_test.cc

@@ -889,8 +889,8 @@ FIXTURE_TEST(oc_ns_adl_serde_no_upgrade, rpc_integration_fixture) {
     start_server();
 
     rpc::transport t(client_config());
-    t.set_version(rpc::transport_version::v0);
     t.connect(model::no_timeout).get();
+    t.set_version(rpc::transport_version::v0); // connect resets version=v1
     auto stop = ss::defer([&t] { t.stop().get(); });
     auto client = echo::echo_client_protocol(t);
 
@@ -920,8 +920,8 @@ FIXTURE_TEST(oc_ns_adl_only_no_upgrade, rpc_integration_fixture) {
     start_server();
 
     rpc::transport t(client_config());
-    t.set_version(rpc::transport_version::v0);
     t.connect(model::no_timeout).get();
+    t.set_version(rpc::transport_version::v0); // connect resets version=v1
     auto stop = ss::defer([&t] { t.stop().get(); });
     auto client = echo::echo_client_protocol(t);
 

src/v/rpc/transport.cc

@@ -83,6 +83,7 @@ void transport::reset_state() {
     _correlation_idx = 0;
     _last_seq = sequence_t{0};
     _seq = sequence_t{0};
+    _version = transport_version::v1;
 }
 
 ss::future<>

src/v/rpc/transport.h

@@ -112,6 +112,8 @@ class transport final : public net::base_transport {
      * version level used when dispatching requests. this value may change
      * during the lifetime of the transport. for example the version may be
      * upgraded if it is discovered that a server supports a newer version.
+     *
+     * reset to v1 in reset_state() to support reconnect_transport.
      */
     transport_version _version{transport_version::v1};