quiche: support connection ID longer than 8 bytes with multiple worker threads

source/common/quic/envoy_quic_dispatcher.cc

@@ -1,5 +1,6 @@
 #include "common/quic/envoy_quic_dispatcher.h"
 
+#include "common/common/safe_memcpy.h"
 #include "common/http/utility.h"
 #include "common/quic/envoy_quic_server_connection.h"
 #include "common/quic/envoy_quic_server_session.h"
@@ -92,5 +93,18 @@ std::unique_ptr<quic::QuicSession> EnvoyQuicDispatcher::CreateQuicSession(
   return quic_session;
 }
 
+quic::QuicConnectionId EnvoyQuicDispatcher::ReplaceLongServerConnectionId(
+    const quic::ParsedQuicVersion& version, const quic::QuicConnectionId& server_connection_id,
+    uint8_t expected_server_connection_id_length) const {
+  quic::QuicConnectionId new_connection_id = quic::QuicDispatcher::ReplaceLongServerConnectionId(
+      version, server_connection_id, expected_server_connection_id_length);
+  char* new_connection_id_data = new_connection_id.mutable_data();
+  const char* server_connection_id_ptr = server_connection_id.data();
+  auto* first_four_bytes = reinterpret_cast<const uint32_t*>(server_connection_id_ptr);
+  // Override the first 4 bytes of the new CID to the original CID's first 4 bytes.
+  safeMemcpyUnsafeDst(new_connection_id_data, first_four_bytes);
+  return new_connection_id;
+}
+
 } // namespace Quic
 } // namespace Envoy

source/common/quic/envoy_quic_dispatcher.h

@@ -60,12 +60,19 @@ class EnvoyQuicDispatcher : public quic::QuicDispatcher {
                           quic::ConnectionCloseSource source) override;
 
 protected:
+  // quic::QuicDispatcher
   std::unique_ptr<quic::QuicSession> CreateQuicSession(quic::QuicConnectionId server_connection_id,
                                                        const quic::QuicSocketAddress& self_address,
                                                        const quic::QuicSocketAddress& peer_address,
                                                        absl::string_view alpn,
                                                        const quic::ParsedQuicVersion& version,
                                                        absl::string_view sni) override;
+  // Overridden to restore the first 4 bytes of the connection ID because our BPF filter only looks
+  // at the first 4 bytes. This ensures that the replacement routes to the same quic dispatcher.
+  quic::QuicConnectionId
+  ReplaceLongServerConnectionId(const quic::ParsedQuicVersion& version,
+                                const quic::QuicConnectionId& server_connection_id,
+                                uint8_t expected_server_connection_id_length) const override;
 
 private:
   Network::ConnectionHandler& connection_handler_;

test/integration/quic_http_integration_test.cc

@@ -212,7 +212,7 @@ class QuicHttpIntegrationTest : public HttpIntegrationTest, public QuicMultiVers
     std::vector<IntegrationCodecClientPtr> codec_clients;
     for (size_t i = 1; i <= concurrency_; ++i) {
       // The BPF filter and ActiveQuicListener::destination() look at the 1st word of connection id
-      // in the packet header. And currently all QUIC versions support 8 bytes connection id. So
+      // in the packet header. And currently all QUIC versions support >= 8 bytes connection id. So
       // create connections with the first 4 bytes of connection id different from each
       // other so they should be evenly distributed.
       designated_connection_ids_.push_back(quic::test::TestConnectionId(i << 32));
@@ -245,6 +245,24 @@ class QuicHttpIntegrationTest : public HttpIntegrationTest, public QuicMultiVers
       }
     }
     for (size_t i = 0; i < concurrency_; ++i) {
+      fake_upstream_connection_ = nullptr;
+      upstream_request_ = nullptr;
+      auto encoder_decoder =
+          codec_clients[i]->startRequest(Http::TestRequestHeaderMapImpl{{":method", "GET"},
+                                                                        {":path", "/test/long/url"},
+                                                                        {":scheme", "http"},
+                                                                        {":authority", "host"}});
+      auto& request_encoder = encoder_decoder.first;
+      auto response = std::move(encoder_decoder.second);
+      codec_clients[i]->sendData(request_encoder, 0, true);
+      waitForNextUpstreamRequest();
+      upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"},
+                                                                       {"set-cookie", "foo"},
+                                                                       {"set-cookie", "bar"}},
+                                       true);
+
+      ASSERT_TRUE(response->waitForEndStream());
+      EXPECT_TRUE(response->complete());
       codec_clients[i]->close();
     }
   }
@@ -384,7 +402,18 @@ TEST_P(QuicHttpIntegrationTest, MultipleQuicConnectionsNoBPF) {
   testMultipleQuicConnections();
 }
 
-TEST_P(QuicHttpIntegrationTest, ConnectionMigration) {
+// Tests that the packets from a connection with CID longer than 8 bytes are routed to the same
+// worker.
+TEST_P(QuicHttpIntegrationTest, MultiWorkerWithLongConnectionId) {
+  concurrency_ = 8;
+  set_reuse_port_ = true;
+  initialize();
+  // Setup 9-byte CID for the next connection.
+  designated_connection_ids_.push_back(quic::test::TestConnectionIdNineBytesLong(2u));
+  testRouterHeaderOnlyRequestAndResponse();
+}
+
+TEST_P(QuicHttpIntegrationTest, PortMigration) {
   concurrency_ = 2;
   set_reuse_port_ = true;
   initialize();