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

livekit_api/services/
twirp_client.rs
