livekit_api/services/
twirp_client.rs

1// Copyright 2025 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{fmt::Display, time::Duration};
16
17use http::{
18    header::{HeaderMap, HeaderValue, CONTENT_TYPE, USER_AGENT},
19    StatusCode,
20};
21use serde::Deserialize;
22use thiserror::Error;
23use url::Url;
24
25use super::failover::{self, FailoverConfig};
26use crate::http_client;
27
28pub const DEFAULT_PREFIX: &str = "/twirp";
29
30/// Identifies the SDK and version to the server on every request.
31const USER_AGENT_VALUE: &str = concat!("livekit-server-sdk-rust/", env!("CARGO_PKG_VERSION"));
32
33/// LiveKit URLs are commonly `wss://` (or `ws://`); the server APIs are Twirp
34/// over HTTP, so the scheme is normalized to `https://` (or `http://`).
35fn normalize_host(host: &str) -> String {
36    if let Some(rest) = host.strip_prefix("wss://") {
37        format!("https://{rest}")
38    } else if let Some(rest) = host.strip_prefix("ws://") {
39        format!("http://{rest}")
40    } else {
41        host.to_owned()
42    }
43}
44
45#[cfg(test)]
46mod normalize_host_tests {
47    use super::normalize_host;
48
49    #[test]
50    fn normalizes_ws_schemes() {
51        assert_eq!(normalize_host("wss://my.livekit.cloud"), "https://my.livekit.cloud");
52        assert_eq!(normalize_host("ws://localhost:7880"), "http://localhost:7880");
53        assert_eq!(normalize_host("https://my.livekit.cloud"), "https://my.livekit.cloud");
54        assert_eq!(normalize_host("http://localhost:7880"), "http://localhost:7880");
55    }
56}
57
58#[derive(Debug, Error)]
59pub enum ServerError {
60    #[error("failed to execute the request: {0}")]
61    Request(#[from] reqwest::Error),
62    #[error("server error: {0}")]
63    Twirp(ServerErrorCode),
64    #[error("url error: {0}")]
65    Url(#[from] url::ParseError),
66    #[error("prost error: {0}")]
67    Prost(#[from] prost::DecodeError),
68}
69
70/// Deprecated alias for [`ServerError`], kept for backwards compatibility.
71pub type TwirpError = ServerError;
72
73#[derive(Debug, Deserialize)]
74pub struct ServerErrorCode {
75    pub code: String,
76    pub msg: String,
77    /// Extra key/value context the server attached (e.g. SIP status). Absent on
78    /// most errors.
79    #[serde(default)]
80    pub meta: std::collections::HashMap<String, String>,
81}
82
83impl ServerErrorCode {
84    pub const CANCELED: &'static str = "canceled";
85    pub const UNKNOWN: &'static str = "unknown";
86    pub const INVALID_ARGUMENT: &'static str = "invalid_argument";
87    pub const MALFORMED: &'static str = "malformed";
88    pub const DEADLINE_EXCEEDED: &'static str = "deadline_exceeded";
89    pub const NOT_FOUND: &'static str = "not_found";
90    pub const BAD_ROUTE: &'static str = "bad_route";
91    pub const ALREADY_EXISTS: &'static str = "already_exists";
92    pub const PERMISSION_DENIED: &'static str = "permission_denied";
93    pub const UNAUTHENTICATED: &'static str = "unauthenticated";
94    pub const RESOURCE_EXHAUSTED: &'static str = "resource_exhausted";
95    pub const FAILED_PRECONDITION: &'static str = "failed_precondition";
96    pub const ABORTED: &'static str = "aborted";
97    pub const OUT_OF_RANGE: &'static str = "out_of_range";
98    pub const UNIMPLEMENTED: &'static str = "unimplemented";
99    pub const INTERNAL: &'static str = "internal";
100    pub const UNAVAILABLE: &'static str = "unavailable";
101    pub const DATA_LOSS: &'static str = "dataloss";
102}
103
104impl Display for ServerErrorCode {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        write!(f, "{}: {}", self.code, self.msg)
107    }
108}
109
110pub type ServerResult<T> = Result<T, ServerError>;
111
112/// Deprecated alias for [`ServerErrorCode`], kept for backwards compatibility.
113pub type TwirpErrorCode = ServerErrorCode;
114
115/// Deprecated alias for [`ServerResult`], kept for backwards compatibility.
116pub type TwirpResult<T> = ServerResult<T>;
117
118#[derive(Debug)]
119pub struct TwirpClient {
120    host: String,
121    pkg: String,
122    prefix: String,
123    client: http_client::Client,
124    failover: FailoverConfig,
125    request_timeout: Duration,
126    // Headers added to every request; used by tests to inject mock directives
127    // since the public service methods don't expose per-call headers.
128    #[cfg(test)]
129    default_headers: HeaderMap,
130}
131
132impl TwirpClient {
133    pub fn new(host: &str, pkg: &str, prefix: Option<&str>) -> Self {
134        Self::with_client(host, pkg, prefix, http_client::Client::new())
135    }
136
137    /// Like [`new`](Self::new) but reuses an existing HTTP client (and its
138    /// connection pool) instead of creating one — the unified [`LiveKitApi`]
139    /// builds one client and shares it across all its services this way.
140    ///
141    /// [`LiveKitApi`]: super::LiveKitApi
142    pub(crate) fn with_client(
143        host: &str,
144        pkg: &str,
145        prefix: Option<&str>,
146        client: http_client::Client,
147    ) -> Self {
148        Self {
149            host: normalize_host(host),
150            pkg: pkg.to_owned(),
151            prefix: prefix.unwrap_or(DEFAULT_PREFIX).to_owned(),
152            client,
153            failover: FailoverConfig::default(),
154            request_timeout: failover::DEFAULT_REQUEST_TIMEOUT,
155            #[cfg(test)]
156            default_headers: HeaderMap::new(),
157        }
158    }
159
160    #[cfg(test)]
161    pub(crate) fn with_default_headers(mut self, headers: HeaderMap) -> Self {
162        self.default_headers = headers;
163        self
164    }
165
166    /// Enables or disables region failover (enabled by default). Failover only
167    /// engages for LiveKit Cloud hosts.
168    pub fn with_failover(mut self, enabled: bool) -> Self {
169        self.failover.enabled = enabled;
170        self
171    }
172
173    /// Overrides the default per-attempt request timeout (10s) applied to calls
174    /// that don't pass their own. Each failover attempt gets the full budget.
175    pub fn with_request_timeout(mut self, timeout: Duration) -> Self {
176        self.request_timeout = timeout;
177        self
178    }
179
180    /// Overrides the full failover configuration, including the internal
181    /// test-only `force` and `backoff_base` knobs.
182    #[cfg(test)]
183    pub(crate) fn with_failover_config(mut self, config: FailoverConfig) -> Self {
184        self.failover = config;
185        self
186    }
187
188    /// Issues a Twirp request, failing over to alternative regions on retryable
189    /// errors. On any transport error or HTTP 5xx it discovers regions via
190    /// `/settings/regions` and replays the request — body and headers intact —
191    /// against the next untried region, with exponential backoff. A 4xx is
192    /// returned immediately.
193    pub async fn request<D: prost::Message, R: prost::Message + Default>(
194        &self,
195        service: &str,
196        method: &str,
197        data: D,
198        headers: HeaderMap,
199    ) -> ServerResult<R> {
200        self.request_with_timeout(service, method, data, headers, self.request_timeout).await
201    }
202
203    /// Like [`request`](Self::request) but with an explicit per-attempt timeout,
204    /// for calls (e.g. SIP dialing) that need a longer budget than the default.
205    pub async fn request_with_timeout<D: prost::Message, R: prost::Message + Default>(
206        &self,
207        service: &str,
208        method: &str,
209        data: D,
210        mut headers: HeaderMap,
211        timeout: Duration,
212    ) -> ServerResult<R> {
213        let original = Url::parse(&self.host)?;
214        let path = format!("{}/{}.{}/{}", self.prefix, self.pkg, service, method);
215        headers.insert(USER_AGENT, HeaderValue::from_static(USER_AGENT_VALUE));
216        #[cfg(test)]
217        for (k, v) in &self.default_headers {
218            headers.insert(k.clone(), v.clone());
219        }
220        let forward = headers.clone(); // headers for the discovery fetch (no content-type yet)
221        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/protobuf"));
222        let body = data.encode_to_vec();
223
224        let max_attempts = self.failover.attempts(original.host_str(), timeout);
225        let mut attempted = vec![failover::host_key(&original)];
226        let mut region_urls: Option<Vec<String>> = None;
227        let mut current = original.clone();
228
229        for attempt in 0..max_attempts {
230            let is_last = attempt + 1 >= max_attempts;
231            let mut url = current.clone();
232            url.set_path(&path);
233
234            let send = self
235                .client
236                .post(url)
237                .headers(headers.clone())
238                .body(body.clone())
239                .timeout(timeout)
240                .send()
241                .await;
242            // The next untried region to fail over to, and a description of the
243            // failure for logging. `None` next means give up and surface the error.
244            let (next, reason) = match send {
245                Ok(resp) => {
246                    let status = resp.status();
247                    if status == StatusCode::OK {
248                        return Ok(R::decode(resp.bytes().await?)?);
249                    }
250                    // 4xx is terminal; only 5xx is retryable.
251                    let next = if is_last || status.as_u16() < 500 {
252                        None
253                    } else {
254                        self.next_region(&original, &forward, &mut region_urls, &attempted).await
255                    };
256                    // No fallback: surface the server's error (needs the body).
257                    let Some(next) = next else {
258                        let err: ServerErrorCode = resp.json().await?;
259                        return Err(ServerError::Twirp(err));
260                    };
261                    drop(resp); // release the connection before backing off
262                    (next, format!("status {status}"))
263                }
264                Err(err) => {
265                    let next = if is_last {
266                        None
267                    } else {
268                        self.next_region(&original, &forward, &mut region_urls, &attempted).await
269                    };
270                    match next {
271                        Some(next) => (next, err.to_string()),
272                        None => return Err(err.into()),
273                    }
274                }
275            };
276
277            log::warn!(
278                "livekit API request to {} failed ({}), retrying with fallback url {}",
279                current.host_str().unwrap_or_default(),
280                reason,
281                next,
282            );
283            failover::backoff_sleep(self.backoff(attempt)).await;
284            attempted.push(failover::host_key(&next));
285            current = next;
286        }
287        unreachable!("failover loop always returns within the attempt budget")
288    }
289
290    fn backoff(&self, attempt: u32) -> std::time::Duration {
291        self.failover.backoff_base * (1u32 << attempt)
292    }
293
294    /// Resolves the next untried region, fetching the region list lazily on the
295    /// first retryable failure and reusing it thereafter.
296    async fn next_region(
297        &self,
298        original: &Url,
299        forward: &HeaderMap,
300        region_urls: &mut Option<Vec<String>>,
301        attempted: &[String],
302    ) -> Option<Url> {
303        let region_urls = match region_urls {
304            Some(urls) => urls,
305            None => region_urls.insert(failover::region_urls(original, forward).await),
306        };
307        failover::pick_next(region_urls, attempted)
308    }
309}
livekit_api/services/twirp_client.rs

livekit_api/services/
twirp_client.rs
