feat(grpc): Add resolver wrapper to handle HTTP CONNECT proxy configuration

grpc/Cargo.toml

@@ -57,8 +57,10 @@ hickory-resolver = { version = "0.25.1", optional = true }
 http = "1.1.0"
 http-body = "1.0.1"
 hyper = { version = "1.6.0", features = ["client", "http2"] }
+hyper-util = { version = "0.1", features = ["client", "client-proxy"] }
 itoa = "1.0"
 parking_lot = "0.12.4"
+percent-encoding = "2.3"
 pin-project-lite = "0.2.16"
 rand = "0.9"
 rustls = { version = "0.23", optional = true, default-features = false, features = [

grpc/src/byte_str.rs

@@ -21,37 +21,111 @@
  * IN THE SOFTWARE.
  *
  */
-
-use core::str;
 use std::ops::Deref;
 
 use bytes::Bytes;
 
 /// A cheaply cloneable and sliceable chunk of contiguous memory.
+///
+/// The bytes held by `ByteStr` are arbitrary and may not be valid UTF-8.
 #[derive(Debug, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct ByteStr {
-    // Invariant: bytes contains valid UTF-8
     bytes: Bytes,
 }
 
+impl ByteStr {
+    /// Strips a prefix, returning a new zero-copy ByteStr.
+    #[inline]
+    pub(crate) fn strip_prefix(&self, prefix: &[u8]) -> Option<ByteStr> {
+        if self.starts_with(prefix) {
+            Some(ByteStr {
+                bytes: self.bytes.slice(prefix.len()..),
+            })
+        } else {
+            None
+        }
+    }
+}
+
 impl Deref for ByteStr {
-    type Target = str;
+    type Target = [u8];
 
     #[inline]
-    fn deref(&self) -> &str {
-        let b: &[u8] = self.bytes.as_ref();
-        // The invariant of `bytes` is that it contains valid UTF-8 allows us
-        // to unwrap.
-        str::from_utf8(b).unwrap()
+    fn deref(&self) -> &[u8] {
+        &self.bytes
     }
 }
 
 impl From<String> for ByteStr {
     #[inline]
     fn from(src: String) -> ByteStr {
         ByteStr {
-            // Invariant: src is a String so contains valid UTF-8.
             bytes: Bytes::from(src),
         }
     }
 }
+
+impl<'a> TryFrom<&'a ByteStr> for &'a str {
+    type Error = std::str::Utf8Error;
+
+    #[inline]
+    fn try_from(value: &'a ByteStr) -> Result<Self, Self::Error> {
+        std::str::from_utf8(value)
+    }
+}
+
+impl TryFrom<ByteStr> for String {
+    type Error = std::str::Utf8Error;
+
+    #[inline]
+    fn try_from(value: ByteStr) -> Result<Self, Self::Error> {
+        let s = std::str::from_utf8(&value)?;
+        Ok(s.to_owned())
+    }
+}
+
+impl PartialEq<str> for ByteStr {
+    #[inline]
+    fn eq(&self, other: &str) -> bool {
+        self.bytes == other.as_bytes()
+    }
+}
+
+impl<'a> PartialEq<&'a str> for ByteStr {
+    #[inline]
+    fn eq(&self, other: &&'a str) -> bool {
+        self.bytes == other.as_bytes()
+    }
+}
+
+impl PartialEq<ByteStr> for str {
+    #[inline]
+    fn eq(&self, other: &ByteStr) -> bool {
+        self.as_bytes() == other.bytes
+    }
+}
+
+impl PartialEq<ByteStr> for &str {
+    #[inline]
+    fn eq(&self, other: &ByteStr) -> bool {
+        self.as_bytes() == other.bytes
+    }
+}
+
+impl FromIterator<u8> for ByteStr {
+    #[inline]
+    fn from_iter<T: IntoIterator<Item = u8>>(iter: T) -> Self {
+        ByteStr {
+            bytes: Bytes::from_iter(iter),
+        }
+    }
+}
+
+impl From<&'static str> for ByteStr {
+    #[inline]
+    fn from(src: &'static str) -> ByteStr {
+        ByteStr {
+            bytes: Bytes::from_static(src.as_bytes()),
+        }
+    }
+}

grpc/src/client/channel.rs

@@ -601,8 +601,8 @@ impl<T: Clone> WatcherIter<T> {
     }
 }
 
-/// Parses the host and port from a URL-encoded string. When the input can not
-/// be parsed as (host, port) pair, it returns the entire input as the host.
+/// Parses the host and port from a string. When the input can not be parsed
+/// as (host, port) pair, it returns the entire input as the host.
 fn parse_authority(host_and_port: &str) -> Authority {
     // Handle bracketed IPv6 addresses (e.g., "[::1]:80").
     if let Some(stripped) = host_and_port.strip_prefix('[')

grpc/src/client/load_balancing/graceful_switch.rs

@@ -452,12 +452,10 @@ mod test {
         let PickResult::Pick(pick) = pick else {
             panic!("unexpected pick result: {:?}", pick);
         };
-        let received_address = &pick.subchannel.address().address.to_string();
-        // It's good practice to create the expected value once.
-        let expected_address = name.to_string();
+        let received_address = &pick.subchannel.address().address;
 
         // Check for inequality and panic with a detailed message if they don't match.
-        assert_eq!(received_address, &expected_address);
+        assert_eq!(received_address, name);
     }
 
     fn move_subchannel_to_state(
@@ -656,7 +654,7 @@ mod test {
             .resolver_update(update.clone(), Some(&parsed_config2), &mut *tcc)
             .unwrap();
         let subchannel = verify_subchannel_creation_from_policy(&mut rx_events);
-        assert_eq!(&*subchannel.address().address, "127.0.0.1:1234");
+        assert_eq!(subchannel.address().address, "127.0.0.1:1234");
         assert_channel_empty(&mut rx_events);
     }
 

grpc/src/client/load_balancing/pick_first.rs

@@ -425,7 +425,7 @@ impl PickFirstPolicy {
         // Partition by family (Basic IPv6 detection via colon).
         let (ipv6, ipv4): (Vec<Address>, Vec<Address>) = tcp_addresses
             .into_iter()
-            .partition(|addr| addr.address.contains(':'));
+            .partition(|addr| addr.address.contains(&b':'));
 
         // Interleave the two lists so ipv6 and ipv4 addresses are alternated.
         let mut interleaved = Vec::with_capacity(ipv6.len() + ipv4.len() + unknown.len());
@@ -738,9 +738,9 @@ mod test {
     use std::time::Duration;
 
     use super::*;
-    use crate::client::load_balancing::test_utils::{
-        TestChannelController, TestEvent, TestWorkScheduler,
-    };
+    use crate::client::load_balancing::test_utils::TestChannelController;
+    use crate::client::load_balancing::test_utils::TestEvent;
+    use crate::client::load_balancing::test_utils::TestWorkScheduler;
 
     const DEFAULT_TEST_DURATION: Duration = Duration::from_secs(10);
 
@@ -928,7 +928,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address.to_string(), "addr1")
+                assert_eq!(pick.subchannel.address().address, "addr1")
             }
             other => panic!("unexpected pick result {:?}", other),
         }
@@ -942,7 +942,7 @@ mod test {
 
         // Should connect to addr2.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr2");
+        assert_eq!(addr.address, "addr2");
 
         // Simulate addr2 succeeding.
         let sc2 = policy.subchannels[1].clone();
@@ -986,25 +986,16 @@ mod test {
 
         // Should create new subchannel for addr2 (was cleared by cleanup).
         let sc2 = expect_new_subchannel(&rx);
-        assert_eq!(sc2.address().address.to_string(), "addr2");
+        assert_eq!(sc2.address().address, "addr2");
         // Should create new subchannel for addr3 (was not in previous list).
         let sc3 = expect_new_subchannel(&rx);
-        assert_eq!(sc3.address().address.to_string(), "addr3");
+        assert_eq!(sc3.address().address, "addr3");
 
         // Should NOT have any more events (no Connect, no UpdatePicker),
         // because it stuck to the original selected subchannel.
         assert!(rx.try_recv().is_err(), "unexpected event");
 
-        assert_eq!(
-            policy
-                .selected
-                .as_ref()
-                .unwrap()
-                .address()
-                .address
-                .to_string(),
-            "addr1"
-        );
+        assert_eq!(policy.selected.as_ref().unwrap().address().address, "addr1");
     }
 
     // If all addresses fail during a connection pass, the LB should update to
@@ -1088,7 +1079,7 @@ mod test {
         let mut resulting_addrs = Vec::with_capacity(NUM_ADDRS);
         for _ in 0..NUM_ADDRS {
             let sc = expect_new_subchannel(&rx);
-            resulting_addrs.push(sc.address().address.to_string());
+            resulting_addrs.push(sc.address().address);
         }
 
         // Mock shuffler reverses endpoints: [EP3, EP2, EP1]
@@ -1157,9 +1148,9 @@ mod test {
 
         // Should only create subchannels for addr1 and addr2 (2 unique addrs).
         let sc1 = expect_new_subchannel(&rx);
-        assert_eq!(sc1.address().address.to_string(), "addr1");
+        assert_eq!(sc1.address().address, "addr1");
         let sc2 = expect_new_subchannel(&rx);
-        assert_eq!(sc2.address().address.to_string(), "addr2");
+        assert_eq!(sc2.address().address, "addr2");
 
         // Verify no 3rd subchannel was created.
         while let Ok(event) = rx.try_recv() {
@@ -1226,7 +1217,7 @@ mod test {
 
         // Expect Connect event for addr2 due to timer expiration.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr2");
+        assert_eq!(addr.address, "addr2");
     }
 
     // If all addresses fail during a connection pass, the LB should enter
@@ -1259,7 +1250,7 @@ mod test {
 
         // Should automatically call connect() again.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr1");
+        assert_eq!(addr.address, "addr1");
     }
 
     // If the LB is in steady state, and a new address becomes ready, it should
@@ -1274,7 +1265,7 @@ mod test {
 
         // Should failover to addr2: expect Connect(addr2).
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr2");
+        assert_eq!(addr.address, "addr2");
 
         // While addr2 is connecting, simulate addr1 going IDLE (backoff over).
         policy.subchannel_update(
@@ -1309,7 +1300,7 @@ mod test {
         );
         expect_request_resolution(&rx);
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr1");
+        assert_eq!(addr.address, "addr1");
 
         // Confirm LB is in steady state.
         assert!(policy.steady_state.is_some());
@@ -1326,7 +1317,7 @@ mod test {
 
         // Now it should automatically call connect() again.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr1");
+        assert_eq!(addr.address, "addr1");
 
         // Simulate addr1 successfully connecting and becoming READY.
         policy.subchannel_update(
@@ -1344,7 +1335,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address.to_string(), "addr1");
+                assert_eq!(pick.subchannel.address().address, "addr1");
             }
             other => panic!("unexpected pick result {:?}", other),
         }
@@ -1363,7 +1354,7 @@ mod test {
 
         // Expect Connect(addr2).
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr2");
+        assert_eq!(addr.address, "addr2");
 
         // Simulate addr1 backing off and transitioning to IDLE early
         // (while addr2 is still connecting).
@@ -1402,7 +1393,7 @@ mod test {
         // Expect an immediate Connect(addr1) event triggered by the exhaustion
         // loop sweeping up the early IDLE subchannel.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr1");
+        assert_eq!(addr.address, "addr1");
     }
 
     // This test is meant to validate that if a new address with different
@@ -1501,7 +1492,7 @@ mod test {
         policy.work(None, controller.as_mut());
 
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr2");
+        assert_eq!(addr.address, "addr2");
 
         // 2. Simulate addr2 failing first while addr1 is still in flight.
         let sc2 = policy.subchannels[1].clone();
@@ -1583,7 +1574,7 @@ mod test {
         let res = state.picker.pick(&RequestHeaders::default());
         let sc1 = match res {
             PickResult::Pick(pick) => {
-                assert_eq!(pick.subchannel.address().address.to_string(), "addr1");
+                assert_eq!(pick.subchannel.address().address, "addr1");
                 pick.subchannel
             }
             other => panic!("unexpected pick result {:?}", other),
@@ -1619,7 +1610,7 @@ mod test {
 
         // 7. Verify that the policy initiates a reconnection to addr1.
         let addr = expect_connect(&rx);
-        assert_eq!(addr.address.to_string(), "addr1");
+        assert_eq!(addr.address, "addr1");
 
         // And the picker goes to Connecting.
         let state = expect_picker_update(&rx);

grpc/src/client/load_balancing/round_robin.rs

@@ -1018,11 +1018,11 @@ mod test {
         let all_subchannels = verify_subchannel_creation(&mut rx_events, 2);
         let subchannel_one = all_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_one".to_string().into())
+            .find(|sc| sc.address().address == "subchannel_one")
             .unwrap();
         let subchannel_two = all_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_two".to_string().into())
+            .find(|sc| sc.address().address == "subchannel_two")
             .unwrap();
 
         move_subchannel_to_state(
@@ -1081,11 +1081,11 @@ mod test {
         let new_subchannels = verify_subchannel_creation(&mut rx_events, 2);
         let new_sc = new_subchannels
             .iter()
-            .find(|sc| sc.address().address == "new".to_string().into())
+            .find(|sc| sc.address().address == "new")
             .unwrap();
         let old_sc = new_subchannels
             .iter()
-            .find(|sc| sc.address().address == "subchannel_two".to_string().into())
+            .find(|sc| sc.address().address == "subchannel_two")
             .unwrap();
 
         move_subchannel_to_state(