Support http2 upstream phase3

Makefile

@@ -183,7 +183,7 @@ OBSERVABILITY_HEADERS = $(LIB_DIR)/observability/common.h $(LIB_DIR)/observabili
 HTTP2_HEADERS = $(LIB_DIR)/http2/http2_callbacks.h $(LIB_DIR)/http2/http2_connection_handler.h $(LIB_DIR)/http2/http2_constants.h $(LIB_DIR)/http2/http2_session.h $(LIB_DIR)/http2/http2_stream.h $(LIB_DIR)/http2/protocol_detector.h
 WS_HEADERS = $(LIB_DIR)/ws/websocket_connection.h $(LIB_DIR)/ws/websocket_frame.h $(LIB_DIR)/ws/websocket_handshake.h $(LIB_DIR)/ws/websocket_parser.h $(LIB_DIR)/ws/utf8_validate.h
 TLS_HEADERS = $(LIB_DIR)/tls/tls_context.h $(LIB_DIR)/tls/tls_connection.h $(LIB_DIR)/tls/tls_client_context.h
-UPSTREAM_HEADERS = $(LIB_DIR)/upstream/upstream_manager.h $(LIB_DIR)/upstream/upstream_host_pool.h $(LIB_DIR)/upstream/pool_partition.h $(LIB_DIR)/upstream/upstream_connection.h $(LIB_DIR)/upstream/upstream_lease.h $(LIB_DIR)/upstream/upstream_codec.h $(LIB_DIR)/upstream/upstream_http_codec.h $(LIB_DIR)/upstream/upstream_h2_codec.h $(LIB_DIR)/upstream/upstream_h2_stream.h $(LIB_DIR)/upstream/upstream_h2_connection.h $(LIB_DIR)/upstream/h2_connection_table.h $(LIB_DIR)/upstream/h2_settings.h $(LIB_DIR)/upstream/http_request_serializer.h $(LIB_DIR)/upstream/header_rewriter.h $(LIB_DIR)/upstream/retry_policy.h $(LIB_DIR)/upstream/proxy_transaction.h $(LIB_DIR)/upstream/proxy_handler.h $(LIB_DIR)/upstream/upstream_response.h $(LIB_DIR)/upstream/upstream_callbacks.h
+UPSTREAM_HEADERS = $(LIB_DIR)/upstream/upstream_manager.h $(LIB_DIR)/upstream/upstream_host_pool.h $(LIB_DIR)/upstream/pool_partition.h $(LIB_DIR)/upstream/upstream_connection.h $(LIB_DIR)/upstream/upstream_lease.h $(LIB_DIR)/upstream/upstream_codec.h $(LIB_DIR)/upstream/upstream_http_codec.h $(LIB_DIR)/upstream/upstream_h2_codec.h $(LIB_DIR)/upstream/upstream_h2_stream.h $(LIB_DIR)/upstream/upstream_h2_connection.h $(LIB_DIR)/upstream/h2_connection_table.h $(LIB_DIR)/upstream/host_port_key.h $(LIB_DIR)/upstream/h2_settings.h $(LIB_DIR)/upstream/http_request_serializer.h $(LIB_DIR)/upstream/header_rewriter.h $(LIB_DIR)/upstream/retry_policy.h $(LIB_DIR)/upstream/proxy_transaction.h $(LIB_DIR)/upstream/proxy_handler.h $(LIB_DIR)/upstream/upstream_response.h $(LIB_DIR)/upstream/upstream_callbacks.h
 RATE_LIMIT_HEADERS = $(LIB_DIR)/rate_limit/token_bucket.h $(LIB_DIR)/rate_limit/rate_limit_zone.h $(LIB_DIR)/rate_limit/rate_limiter.h
 CIRCUIT_BREAKER_HEADERS = $(LIB_DIR)/circuit_breaker/circuit_breaker_state.h $(LIB_DIR)/circuit_breaker/circuit_breaker_window.h $(LIB_DIR)/circuit_breaker/circuit_breaker_slice.h $(LIB_DIR)/circuit_breaker/retry_budget.h $(LIB_DIR)/circuit_breaker/circuit_breaker_host.h $(LIB_DIR)/circuit_breaker/circuit_breaker_manager.h
 # Auth headers. The vendored jwt-cpp headers are pulled into the dependency

include/upstream/h2_connection_table.h

@@ -45,22 +45,39 @@ class H2ConnectionTable {
 
     // Find the first usable H2 connection for `upstream_name`. Reaps
     // drained entries inline as a side benefit. Returns null when no
-    // tracked connection can host a new stream right now.
-    std::shared_ptr<UpstreamH2Connection> FindUsable(
-        const std::string& upstream_name);
+    // tracked connection can host a new stream right now. Lifetime is
+    // owned by the table; callers that need destroy-safety capture
+    // `conn->alive_token()` alongside the raw pointer.
+    UpstreamH2Connection* FindUsable(const std::string& upstream_name);
 
     // Append a freshly Init()'d connection. Caller has already donated
     // the lease via UpstreamH2Connection::AdoptLease.
     void Insert(const std::string& upstream_name,
-                std::shared_ptr<UpstreamH2Connection> conn);
+                std::unique_ptr<UpstreamH2Connection> conn);
 
     // Run the per-connection liveness Tick on every tracked connection.
     // Connections whose Tick returns false (PING timeout, session-fatal
     // error, GOAWAY drained) get FailAllStreams + erased. Caller must
     // already be on the partition's owning dispatcher thread.
     void TickAll(std::chrono::steady_clock::time_point now);
 
+    // Extract the owning unique_ptr for `conn` out of the table.
+    // Returns null if `conn` is not tracked. Used by
+    // PoolPartition::MoveConnToPendingDestroy to hand ownership to the
+    // post-recv-tick destroy stash without invoking the conn's dtor
+    // inline (which would re-enter callbacks from within a recv
+    // callback).
+    std::unique_ptr<UpstreamH2Connection> Extract(UpstreamH2Connection* conn);
+
+    // Move every tracked connection out of the table. Returned vector
+    // owns the conns; the table is left empty. Used by
+    // PoolPartition::InitiateShutdown to retire H2 sessions via
+    // DestroyOnDispatcher on the partition's dispatcher thread so the
+    // donated leases drop and the partition's outstanding_conns_
+    // counter reaches zero before WaitForDrain times out.
+    std::vector<std::unique_ptr<UpstreamH2Connection>> ExtractAll();
+
 private:
     std::unordered_map<std::string,
-        std::vector<std::shared_ptr<UpstreamH2Connection>>> by_upstream_;
+        std::vector<std::unique_ptr<UpstreamH2Connection>>> by_upstream_;
 };

include/upstream/host_port_key.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include <cstddef>
+#include <functional>
+#include <string>
+
+// User-defined key for the H2 connecting-conns stash and replacement-
+// connect dedup set. A struct (not a `std::pair` alias) is required so
+// that specializing `std::hash<HostPortKey>` below depends on a user-
+// defined type — specializing `std::hash` for library types like
+// `std::pair` would be undefined behavior per [namespace.std].
+struct HostPortKey {
+    std::string host;
+    int port;
+
+    bool operator==(const HostPortKey& other) const noexcept {
+        return port == other.port && host == other.host;
+    }
+    bool operator!=(const HostPortKey& other) const noexcept {
+        return !(*this == other);
+    }
+};
+
+namespace std {
+template <>
+struct hash<HostPortKey> {
+    size_t operator()(const HostPortKey& k) const noexcept {
+        size_t h1 = std::hash<std::string>{}(k.host);
+        size_t h2 = std::hash<int>{}(k.port);
+        // boost::hash_combine mixing pattern, widened to 64 bits via
+        // the floor(2^64 / phi) constant — same shape as
+        // boost::hash_combine_impl<64>, just inlined here so we don't
+        // pull boost in.
+        return h1 ^ (h2 + 0x9e3779b97f4a7c15ULL + (h1 << 6) + (h1 >> 2));
+    }
+};
+}  // namespace std