feat: update relay protocol to iroh-relay-v2, add new Health frame, test protocol update mechanism

iroh-relay/src/client.rs

@@ -27,7 +27,7 @@ pub use self::conn::{RecvError, SendError};
 use crate::dns::{DnsError, DnsResolver};
 use crate::{
  KeyCache,
- http::RELAY_PATH,
+ http::{ProtocolVersion, RELAY_PATH},
  protos::{
  handshake,
  relay::{ClientToRelayMsg, RelayToClientMsg},
@@ -207,7 +207,7 @@ impl ClientBuilder {
  use tls::MaybeTlsStreamBuilder;
 
  use crate::{
- http::{CLIENT_AUTH_HEADER, RELAY_PROTOCOL_VERSION},
+ http::CLIENT_AUTH_HEADER,
  protos::{handshake::KeyMaterialClientAuth, relay::MAX_FRAME_SIZE},
  tls::{CaRootsConfig, default_provider},
  };
@@ -255,7 +255,7 @@ impl ClientBuilder {
  })?
  .add_header(
  SEC_WEBSOCKET_PROTOCOL,
- http::HeaderValue::from_static(RELAY_PROTOCOL_VERSION),
+ ProtocolVersion::all_as_header_value(),
  )
  .expect("valid header name and value")
  .limits(tokio_websockets::Limits::default().max_payload_len(Some(MAX_FRAME_SIZE)))
@@ -312,8 +312,6 @@ impl ClientBuilder {
  /// Establishes a new connection to the relay server.
  #[cfg(wasm_browser)]
  pub async fn connect(&self) -> Result<Client, ConnectError> {
- use crate::http::RELAY_PROTOCOL_VERSION;
-
  let mut dial_url = (*self.url).clone();
  dial_url.set_path(RELAY_PATH);
  // The relay URL is exchanged with the http(s) scheme in tickets and similar.
@@ -332,9 +330,11 @@ impl ClientBuilder {
 
  debug!(%dial_url, "Dialing relay by websocket");
 
- let (_, ws_stream) =
- ws_stream_wasm::WsMeta::connect(dial_url.as_str(), Some(vec![RELAY_PROTOCOL_VERSION]))
- .await?;
+ let (_, ws_stream) = ws_stream_wasm::WsMeta::connect(
+ dial_url.as_str(),
+ Some(ProtocolVersion::all().collect()),
+ )
+ .await?;
  let conn = Conn::new(ws_stream, self.key_cache.clone(), &self.secret_key).await?;
 
  event!(

iroh-relay/src/http.rs

@@ -1,6 +1,8 @@
 //! HTTP-specific constants for the relay server and client.
 
-use http::HeaderName;
+use http::{HeaderName, HeaderValue};
+use n0_error::stack_error;
+use strum::VariantArray;
 
 #[cfg(feature = "server")]
 pub(crate) const WEBSOCKET_UPGRADE_PROTOCOL: &str = "websocket";
@@ -13,10 +15,90 @@ pub const RELAY_PATH: &str = "/relay";
 /// The HTTP path under which the relay allows doing latency queries for testing.
 pub const RELAY_PROBE_PATH: &str = "/ping";
 
-/// The websocket sub-protocol version that we currently support.
-///
-/// This is sent as the websocket sub-protocol header `Sec-Websocket-Protocol` from
-/// the client and answered from the server.
-pub const RELAY_PROTOCOL_VERSION: &str = "iroh-relay-v1";
 /// The HTTP header name for relay client authentication
 pub const CLIENT_AUTH_HEADER: HeaderName = HeaderName::from_static("x-iroh-relay-client-auth-v1");
+
+/// The relay protocol version negotiated between client and server.
+///
+/// Sent as the websocket sub-protocol header `Sec-Websocket-Protocol` from
+/// the client. The server picks the best supported version and replies with it.
+///
+/// Variants are ordered by preference (highest first), so the [`Ord`] impl
+/// can be used during negotiation to pick the best version.
+#[derive(
+ Clone,
+ Copy,
+ Debug,
+ PartialEq,
+ Eq,
+ PartialOrd,
+ Ord,
+ Default,
+ strum::VariantArray,
+ strum::EnumString,
+ strum::Display,
+ strum::IntoStaticStr,
+)]
+#[strum(parse_err_ty = UnsupportedRelayProtocolVersion, parse_err_fn = strum_err_fn)]
+// Needs to be ordered with latest version last, so that the `Ord` impl orders by latest version as max.
+pub enum ProtocolVersion {
+ /// Version 1 (before iroh 0.97.0)
+ #[strum(serialize = "iroh-relay-v1")]
+ V1,
+ /// Version 2 (added in iroh 0.97.0)
+ /// - Removed `Health` frame (id 11)
+ /// - Added new `Status` frame (id 13)
+ /// - Changed behavior such that unknown frames are allowed
+ #[default]
+ #[strum(serialize = "iroh-relay-v2")]
+ V2,
+}
+
+impl ProtocolVersion {
+ /// Returns an iterator of all supported protocol version identifiers, in order of preference.
+ pub fn all() -> impl Iterator<Item = &'static str> {
+ Self::VARIANTS
+ .iter()
+ .map(ProtocolVersion::to_str)
+ // We reverse the order so that the latest version comes first:
+ // `Self::VARIANTS` is ordered in definition order, where the latest version comes last
+ // so that the `Ord` derive correctly orderes by "latest is max".
+ .rev()
+ }
+
+ /// Returns a comma-separated string of all supported protocol version identifiers.
+ pub fn all_joined() -> String {
+ Self::all().collect::<Vec<_>>().join(", ")
+ }
+
+ /// Returns all supported protocol versions in a comma-seperated string as an HTTP header value.
+ pub fn all_as_header_value() -> HeaderValue {
+ HeaderValue::from_bytes(Self::all_joined().as_bytes()).expect("valid header name")
+ }
+
+ /// Returns the protocol version identifier string.
+ pub fn to_str(&self) -> &'static str {
+ self.into()
+ }
+
+ /// Tries to parse a [`ProtocolVersion`] from `s`.
+ ///
+ /// Returns `None` if `s` is not a valid protocol version string.
+ pub fn match_from_str(s: &str) -> Option<Self> {
+ Self::try_from(s).ok()
+ }
+
+ /// Returns this protocol version as an HTTP header value.
+ pub fn to_header_value(&self) -> HeaderValue {
+ HeaderValue::from_static(self.to_str())
+ }
+}
+
+/// Error returned when the relay protocol version is not recognized.
+#[stack_error(derive)]
+#[error("Relay protocol version is not supported")]
+pub struct UnsupportedRelayProtocolVersion;
+
+fn strum_err_fn(_item: &str) -> UnsupportedRelayProtocolVersion {
+ UnsupportedRelayProtocolVersion::new()
+}

iroh-relay/src/protos/common.rs

@@ -45,6 +45,8 @@ pub enum FrameType {
  Ping = 9,
  /// 8 byte payload, the contents of ping being replied to
  Pong = 10,
+ /// REMOVED since relay-protocol-v2, use `Self::Status` instead.
+ ///
  /// Sent from server to client to tell the client if their connection is unhealthy somehow.
  /// Contains only UTF-8 bytes.
  Health = 11,
@@ -53,6 +55,15 @@ pub enum FrameType {
  /// Payload is two big endian u32 durations in milliseconds: when to reconnect,
  /// and how long to try total.
  Restarting = 12,
+
+ /// Sent from server to client to declare the connection health state.
+ ///
+ /// Added in `iroh-relay-v2` protocol. May not be sent to `iroh-relay-v1` clients.
+ ///
+ /// Uses a binary-encoded [`HealthStatus`] payload.
+ ///
+ /// [`HealthStatus`]: super::relay::HealthStatus
+ Status = 13,
 }
 
 #[stack_error(derive, add_meta)]

iroh-relay/src/protos/relay.rs

@@ -71,7 +71,8 @@ pub enum Error {
 }
 
 /// The messages that a relay sends to clients or the clients receive from the relay.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, strum::Display)]
+#[non_exhaustive]
 pub enum RelayToClientMsg {
  /// Represents datagrams sent from relays (originally sent to them by another client).
  Datagrams {
@@ -84,15 +85,7 @@ pub enum RelayToClientMsg {
  /// packet but has now disconnected from the relay.
  EndpointGone(EndpointId),
  /// A one-way message from relay to client, declaring the connection health state.
- Health {
- /// If set, is a description of why the connection is unhealthy.
- ///
- /// If `None` means the connection is healthy again.
- ///
- /// The default condition is healthy, so the relay doesn't broadcast a [`RelayToClientMsg::Health`]
- /// until a problem exists.
- problem: String,
- },
+ Status(HealthStatus),
  /// A one-way message from relay to client, advertising that the relay is restarting.
  Restarting {
  /// An advisory duration that the client should wait before attempting to reconnect.
@@ -110,6 +103,66 @@ pub enum RelayToClientMsg {
  /// Reply to a [`ClientToRelayMsg::Ping`] from a client
  /// with the payload sent previously in the ping.
  Pong([u8; 8]),
+
+ // -- Deprecated variants --
+ // We don't use `#[deprecated]` because this would throw warnings for the derived serde impls.
+ /// Removed since relay-protocol-v2:
+ /// A one-way message from relay to client, declaring the connection health state.
+ ///
+ /// Use [`Self::Status`] instead.
+ Health {
+ /// If set, is a description of why the connection is unhealthy.
+ ///
+ /// If `None` means the connection is healthy again.
+ ///
+ /// The default condition is healthy, so the relay doesn't broadcast a [`RelayToClientMsg::Health`]
+ /// until a problem exists.
+ problem: String,
+ },
+}
+
+/// One-way message from server to client indicating issues with the relay connection.
+#[derive(Debug, Clone, PartialEq, Eq, derive_more::Display)]
+#[non_exhaustive]
+pub enum HealthStatus {
+ /// The connection is healthy and recovered from previous problems.
+ #[display("The connection is healthy and has recovered from previous problems")]
+ Healthy,
+ /// Another endpoint connected with the same endpoint id. No more messages will be received.
+ #[display(
+ "Another endpoint connected with the same endpoint id. No more messages will be received."
+ )]
+ SameEndpointIdConnected,
+ /// Placeholder for backwards-compatibility for future new health status variants.
+ #[display("Unsupported health message ({_0})")]
+ Unknown(u8),
+}
+
+impl HealthStatus {
+ #[cfg(feature = "server")]
+ fn write_to<O: BufMut>(&self, mut dst: O) -> O {
+ match self {
+ HealthStatus::Healthy => dst.put_u8(0),
+ HealthStatus::SameEndpointIdConnected => dst.put_u8(1),
+ HealthStatus::Unknown(discriminant) => dst.put_u8(*discriminant),
+ }
+ dst
+ }
+
+ #[cfg(feature = "server")]
+ fn encoded_len(&self) -> usize {
+ 1
+ }
+
+ fn from_bytes(mut bytes: Bytes) -> Result<Self, Error> {
+ ensure!(!bytes.is_empty(), Error::InvalidFrame);
+ let discriminant = bytes.get_u8();
+ match discriminant {
+ 0 => Ok(Self::Healthy),
+ 1 => Ok(Self::SameEndpointIdConnected),
+ n => Ok(Self::Unknown(n)),
+ }
+ }
 }
 
 /// Messages that clients send to relays.
@@ -257,8 +310,9 @@ impl RelayToClientMsg {
  Self::EndpointGone { .. } => FrameType::EndpointGone,
  Self::Ping { .. } => FrameType::Ping,
  Self::Pong { .. } => FrameType::Pong,
- Self::Health { .. } => FrameType::Health,
+ Self::Status { .. } => FrameType::Status,
  Self::Restarting { .. } => FrameType::Restarting,
+ Self::Health { .. } => FrameType::Health,
  }
  }
 
@@ -300,6 +354,9 @@ impl RelayToClientMsg {
  dst.put_u32(reconnect_in.as_millis() as u32);
  dst.put_u32(try_for.as_millis() as u32);
  }
+ Self::Status(status) => {
+ dst = status.write_to(dst);
+ }
  }
  dst
  }
@@ -313,11 +370,12 @@ impl RelayToClientMsg {
  }
  Self::EndpointGone(_) => 32,
  Self::Ping(_) | Self::Pong(_) => 8,
- Self::Health { problem } => problem.len(),
+ Self::Status(status) => status.encoded_len(),
  Self::Restarting { .. } => {
  4 // u32
  + 4 // u32
  }
+ Self::Health { problem } => problem.len(),
  };
  self.typ().encoded_len() + payload_len
  }
@@ -388,6 +446,10 @@ impl RelayToClientMsg {
  try_for,
  }
  }
+ FrameType::Status => {
+ let status = HealthStatus::from_bytes(content)?;
+ Self::Status(status)
+ }
  _ => {
  return Err(e!(Error::InvalidFrameType { frame_type }));
  }
@@ -599,6 +661,11 @@ mod tests {
  .write_to(Vec::new()),
  "0c 00 00 00 0a 00 00 00 14",
  ),
+ (
+ RelayToClientMsg::Status(HealthStatus::SameEndpointIdConnected)
+ .write_to(Vec::new()),
+ "0d 01",
+ ),
  ]);
 
  Ok(())
@@ -719,18 +786,28 @@ mod proptests {
  let endpoint_gone = key().prop_map(RelayToClientMsg::EndpointGone);
  let ping = prop::array::uniform8(any::<u8>()).prop_map(RelayToClientMsg::Ping);
  let pong = prop::array::uniform8(any::<u8>()).prop_map(RelayToClientMsg::Pong);
- let health = ".{0,65536}"
+ let v1health = ".{0,65536}"
  .prop_filter("exceeds MAX_PACKET_SIZE", |s| {
  s.len() < MAX_PACKET_SIZE // a single unicode character can match a regex "." but take up multiple bytes
  })
  .prop_map(|problem| RelayToClientMsg::Health { problem });
+ let health = Just(HealthStatus::SameEndpointIdConnected)
+ .prop_map(RelayToClientMsg::Status);
  let restarting = (any::<u32>(), any::<u32>()).prop_map(|(reconnect_in, try_for)| {
  RelayToClientMsg::Restarting {
  reconnect_in: Duration::from_millis(reconnect_in.into()),
  try_for: Duration::from_millis(try_for.into()),
  }
  });
- prop_oneof![recv_packet, endpoint_gone, ping, pong, health, restarting]
+ prop_oneof![
+ recv_packet,
+ endpoint_gone,
+ ping,
+ pong,
+ v1health,
+ restarting,
+ health
+ ]
  }
 
  fn client_server_frame() -> impl Strategy<Value = ClientToRelayMsg> {